Я использую rabbitMQ, и мне нужно получить случайное количество сообщений в очереди.
Но, к сожалению, при запуске кода предварительная выборка и настройка Qos не работают.
2 сообщения приходят к comsuume одновременно с тем же тегом diliveryTag.
Так что я не могу получить номер сообщения, как я ожидал.
вот мой код:
exports.getMessageFromQueue = async function(queue,numbMessage) {
var exchange = queue + '.exchange';
var result = [];
var countReceive = 0;
var isDone = false;
let channel = await conn.createChannel()
let response = await channel.assertQueue(queue)
await channel.prefetch(1);
await channel.qos(1);
response = await channel.bindQueue(response.queue, exchange, '');
channel.consume(response.queue,getMessage(numbMessage), {noAck: false})
setTimeout(() => {
eventEmitter.emit('consumeDone');
if(channel != null)channel.close();}, timeout)
await new Promise(resolve => eventEmitter.once('consumeDone', ()=>{
console.log("done event");
resolve()
}));
console.log("return");
return result;
// console.log('reading for query finish')
function getMessage(messageCount) {
return msg => {
var content = msg.content.toString();
var secs = content.split('.').length - 1;
result.push(JSON.parse(content));
console.log("deliveryTag:" + msg.fields.deliveryTag)
if (messageCount == deliveryTag) {
console.log(" [x] Done messageCount");
eventEmitter.emit('consumeDone')
channel.cancel(msg.fields.consumerTag)
channel.ack(msg);
channel.close();
channel = null;
}else {
setTimeout(function() {
console.log(" [x] Done");
channel.ack(msg);
}, secs *1000);
}
}
}
}
Кто-нибудь сталкивался с этой проблемой, пожалуйста, помогите!