Node.js中如何从RabbitMQ消费函数返回数据？

我有如下RabbitMQ消费函数：
当在消费回调里用console.log()时，能正常看到队列传来的消息：

Function 1

async function consumeData() {
  try {
    const connection = await amqp.connect("amqp://localhost:5672");
    const channel = await connection.createChannel();
    await channel.assertQueue(queueName);
    let consumedData;

    channel.consume(queueName, (message) => {
      consumedData = message.content.toString();
      console.log(consumedData);
      channel.ack(message);
    });
  } catch (error) {
    console.log("Error", error);
  }
}

但我不想打印数据，想直接返回并使用，于是改成了下面的函数，运行后却无法返回任何数据：

Function 2

async function consumeData() {
  try {
    const connection = await amqp.connect("amqp://localhost:5672");
    const channel = await connection.createChannel();
    await channel.assertQueue(queueName);
    let consumedData;

    channel.consume(queueName, (message) => {
      consumedData = message.content.toString();
      console.log(consumedData);
      channel.ack(message);
    });
    
    return consumedData;
  } catch (error) {
    console.log("Error", error);
  }
}

请问如何从该消费函数中返回数据？

channel.consume()是异步回调模式，函数执行到return consumedData时，回调还没触发，此时consumedData还是未定义状态，所以返回的是undefined。

根据使用场景，有两种处理方式：

1. 单次消费（获取一条消息后结束） #

如果只需要获取一条消息就停止消费，可以用Promise包裹，拿到消息后取消消费者：

async function consumeData() {
  try {
    const connection = await amqp.connect("amqp://localhost:5672");
    const channel = await connection.createChannel();
    await channel.assertQueue(queueName);

    return new Promise((resolve) => {
      const consumerTag = channel.consume(queueName, (message) => {
        const consumedData = message.content.toString();
        channel.ack(message);
        // 取消消费者，避免持续监听
        channel.cancel(consumerTag);
        // 关闭连接（如果不需要保持的话）
        connection.close();
        resolve(consumedData);
      });
    });
  } catch (error) {
    console.log("Error", error);
    throw error; // 抛出错误让调用方处理
  }
}

// 使用方式
consumeData().then(data => {
  console.log("拿到的数据：", data);
}).catch(err => {
  console.error("消费出错：", err);
});

2. 持续消费（监听队列并处理每条消息） #

如果需要持续监听队列，无法直接返回单条数据，建议把处理逻辑放到回调里，或者通过事件通知传递数据：

// 方式1：直接在回调里处理业务逻辑
async function consumeData() {
  try {
    const connection = await amqp.connect("amqp://localhost:5672");
    const channel = await connection.createChannel();
    await channel.assertQueue(queueName);

    channel.consume(queueName, (message) => {
      const consumedData = message.content.toString();
      channel.ack(message);
      // 在这里处理拿到的数据，比如存入数据库、调用API等
      processData(consumedData);
    });
  } catch (error) {
    console.log("Error", error);
  }
}

function processData(data) {
  // 你的业务处理逻辑
  console.log("处理数据：", data);
}

// 方式2：用EventEmitter传递消息
const EventEmitter = require('events');
const messageEmitter = new EventEmitter();

async function consumeData() {
  try {
    const connection = await amqp.connect("amqp://localhost:5672");
    const channel = await connection.createChannel();
    await channel.assertQueue(queueName);

    channel.consume(queueName, (message) => {
      const consumedData = message.content.toString();
      channel.ack(message);
      messageEmitter.emit('new-message', consumedData);
    });
  } catch (error) {
    console.log("Error", error);
  }
}

// 使用方式
messageEmitter.on('new-message', (data) => {
  console.log("收到新消息：", data);
  // 处理数据
});

consumeData();

内容的提问来源于stack exchange，提问作者hdevtr