如何使用b-rabbit库消费RBMQ队列消息并实现消息确认？

b-rabbit并非仅支持消息发布——它基于amqp-lib封装，完全支持消息消费以及消息确认（手动/自动）机制，只是官方文档对消费相关的细节说明较少。以下是具体实现示例：

1. 基本消息消费实现 #

你可以通过subscribe方法创建消费者，监听指定队列：

const BRabbit = require('b-rabbit');

const rabbit = new BRabbit({
  host: 'your-rabbitmq-host',
  port: 5672,
  username: 'guest',
  password: 'guest'
});

// 初始化连接
rabbit.start()
  .then(() => {
    // 订阅队列
    rabbit.subscribe({
      queue: 'your-queue-name',
      exchange: 'your-exchange-name', // 可选，若队列绑定了交换机需指定
      routingKey: 'your-routing-key', // 可选
      onMessage: (msg, channel) => {
        console.log('收到消息:', msg.content.toString());
        // 这里可以处理消息确认逻辑
      }
    });
  })
  .catch(err => console.error('连接失败:', err));

2. 消息确认机制实现 #

b-rabbit支持自动确认和手动确认两种模式：

• 自动确认：默认开启，无需额外操作，但这种模式下RabbitMQ会在消息发送给消费者后立即标记为已消费，若消费者崩溃会丢失消息。
• 手动确认：需在配置中关闭自动确认，处理完消息后手动发送确认信号：

rabbit.subscribe({
  queue: 'your-queue-name',
  options: {
    noAck: false // 关闭自动确认，开启手动确认
  },
  onMessage: (msg, channel) => {
    try {
      // 处理消息逻辑
      console.log('处理消息:', msg.content.toString());
      // 手动确认消息，第二个参数为true表示确认所有未确认的消息
      channel.ack(msg, false);
    } catch (err) {
      console.error('消息处理失败:', err);
      // 拒绝消息，requeue设为true会将消息重新放回队列
      channel.nack(msg, false, true);
      // 或直接丢弃消息：channel.reject(msg, false);
    }
  }
});

额外注意 #

b-rabbit的底层是amqp-lib，而amqp-lib的通道（channel）并非线程安全的——如果你的应用涉及多线程消费，建议为每个线程创建独立的通道或连接，避免并发操作同一通道导致的异常。

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