如何使用lib amqp.node中的方法消耗获取所有消息

Question

问候伙计们。
你能帮我解决节点.js中的异步吗？

问题这个：

我正在使用 amqplib 模块与 RabbitMQ 一起工作，这里有方法消耗，谁从 RabbitMQ 给出消息，但该方法首先返回关于他开始的承诺，在这个承诺开始后，他调用回调从 RabbitMQ 获取数据，我不知道如何捕获何时将所有消息发送到我的节点 js 应用程序。

为了进一步解释，这里是我的代码和注释中的结束代码，我写了我想要的：

/**
 * Here my test code
 *
 * requirng amqp.node lib
 */
let amqp = require('amqplib')
  , configConnection = { /* ..config options */ }
  , queue = 'users'
  , exchange = 'users.exchange'
  , type = 'fanout'
/**
 * declare annonymous function as async and immediately called it
 */
(async () => {
  /**
   * declare connection and channel with using async/await construction
   * who support version node.js >= 8.5.0
   */
  let conn = await amqp.connect(configConnection)
  let channel = await conn.createChannel()
  await channel.assertExchange(exchange, type)
  let response = await channel.assertQueue(queue)
  /**
   * response: { queue: 'users', messageCount: 10, consumerCount: 0 }
   */
  response = await channel.bindQueue(response.queue, exchange, '')
  response = await channel.consume(response.queue, logMessage, {noAck: false})
  /**
   * {noAck: false} false for not expect an acknowledgement
   */
  console.log('reading for query finish')
  function logMessage(msg) {
    console.log("[*] recieved: '%s'", msg.content.toString())
  }
})()
  /**
   * output will show:
   * reading for query finish
   * [*] recieved: 'message content'
   * [*] recieved: 'message content'
   * [*] recieved: 'message content'
   * ...
   *
   * But i'm need show message 'reading for query finish' after when
   * all consumes will executed
   *
   * Ask: How i can do this?
   */

Answer 1

我在这里找到了问题的答案。

使用中的答案：Event发射器&&Promise。

魔术(对我来说(就在这里：
await new Promise(resolve => eventEmitter.once('consumeDone', resolve))

所以结束的代码是：

/**
 * Here my test code
 *
 * requirng amqp.node lib
 */
let amqp = require('amqplib')
  , EventEmitter = require('events')
  , eventEmitter = new EventEmitter()
  , timeout = 10000
  , configConnection = { /* ..config options */ }
  , queue = 'users'
  , exchange = 'users.exchange'
  , type = 'fanout'
/**
 * declare annonymous function as async and immediately called it
 */
(async () => {
  /**
   * declare connection and channel with using async/await construction
   * who support version node.js >= 8.5.0
   */
  let conn = await amqp.connect(configConnection)
  let channel = await conn.createChannel()
  await channel.assertExchange(exchange, type)
  let response = await channel.assertQueue(queue)
  /**
   * response: { queue: 'users', messageCount: 10, consumerCount: 0 }
   */
  let messageCount = response.messageCount
  response = await channel.bindQueue(response.queue, exchange, '')
  response = await channel.consume(response.queue, logMessage(messageCount), {noAck: false})
  /**
   * {noAck: false} false for not expect an acknowledgement
   */
  /**
   * declare timeout if we have problems with emit event in consume
   * we waiting when event will be emit once 'consumeDone' and promise gain resolve
   * so we can go to the next step
   */
  setTimeout(() => eventEmitter.emit('consumeDone'), timeout)
  await new Promise(resolve => eventEmitter.once('consumeDone', resolve))
  console.log('reading for query finish')
  function logMessage(messageCount) {
    return msg => {
      console.log("[*] recieved: '%s'", msg.content.toString())
      if (messageCount == msg.fields.deliveryTag) {
        eventEmitter.emit('consumeDone')
      }
    }
  }
})()

Answer 2

尝试以下代码：

let amqp = require('amqplib/callback_api');
function init() {
    let configConnection = {
        protocol: 'amqp',
        hostname: 'localhost',
        port: 5672,
        username: 'root',
        password: '1111',
        heartbeat: 60,
    };
    let messageCount = 0;
    let messageCounter = 0;
    let queue_name = 'queue_name';
    let messages = [];

    amqp.connect(configConnection, function (error, connect) {
        if (error) {
            throw error;
        }
        // Create channel and get info about queue
        connect.createChannel(function (error1, channel) {
            if (error1) {
                throw error1;
            }
            channel.assertQueue(queue_name, {durable: true}, (error2, result) => {
                // here you get count of messages
                messageCount = result.messageCount;
                // Consume to queue
                channel.consume(queue_name, function (msg) {
                    msg = msg.content.toString();
                    messages.push(msg);
                    // Show all messages and exit
                    if (messageCount === ++messageCounter) {
                        console.log(messages);
                        process.exit();
                    }
                }, {
                    noAck: true
                });
            });
        });
    });
}
init();