RabbitMQ 系列之：消息确认机制（Confirm）

消息的确认机制（Confirm）是实现 RabbitMQ 消息高可用投递的关键步骤，生产者在发送消息之后，如果 Broker 收到消息则会给生产端一个答复，用来确定这条消息是否正常发送。

Confirm 的另一个优点是它所处理的逻辑是异步的，生产者在发送一条消息之后可以进行异步监听通过回调来处理该消息，同时还可继续发送下一条，如果 RabbitMQ 因为系统内部错误导致消息丢失，回调会收到一条 nack 消息，用来处理失败的逻辑，否则会收到一条 ack 成功的消息。

Node.js 版实现

一个 channel 使用确认模式 “confirmation mode”，可参考官方文章 http://www.squaremobius.net/amqp.node/channel_api.html#confirmchannel

实现步骤

  1. 通过 connect.createConfirmChannel() 获取一个确认模式的 channel
  2. 使用 publish() 或者 sendToQueue() 接收回调做为附加参数

代码实现

主要变动生产者代码部分

  1. const amqp = require('amqplib');
  3. async function producer() {
  4.     // 创建链接对象
  5.     const connection = await amqp.connect('amqp://localhost:5672');
  7.     // 获取通道
  8.     const channel = await connection.createConfirmChannel();
  10.     // 声明参数
  11.     const exchangeName = 'confirm_exchange_name';
  12.     const routingKey = 'confirm_routingKey';
  13.     const msg = 'confirm_hello world';
  15.     // 交换机
  16.     await channel.assertExchange(exchangeName, 'direct', {
  17.         durable: true,
  18.     });
  20.     // 发送消息
  21.     channel.publish(exchangeName, routingKey, Buffer.from(msg), {}, function(err, ok){
  22.         console.log(err, ok);
  24.         if (err !== null) {
  25.             console.warn('Message nacked!');
  26.         } else {
  27.             console.log('Message acked');
  28.         }
  29.     });
  31.     // 关闭链接
  32.     // await channel.close();
  33.     // await connection.close();
  34. }
  36. producer();

源码地址

  1. https://github.com/Q-Angelo/project-training/tree/master/rabbitmq/helloworld-confirm

Java 版实现

实现步骤

  1. 开启确认模式 channel.confirmSelect()
  2. 添加监听 channel.addConfirmListener 监听失败或成功结果

代码实现

主要改动部分为生产者部分，并且取消了关闭了链接（channel.close()、connection.close()）否则就无法监听

  1. public class Producer {
  2.     public static void main(String[] args) throws Exception {
  3.         // 创建链接工厂
  4.         ConnectionFactory connectionFactory = new ConnectionFactory();
  5.         connectionFactory.setHost("127.0.0.1");
  6.         connectionFactory.setPort(5672);
  7.         connectionFactory.setVirtualHost("/");
  9.         // 通过链接工厂创建链接
  10.         Connection connection = connectionFactory.newConnection();
  12.         // 通过链接创建通道（channel）
  13.         Channel channel = connection.createChannel();
  15.         // 指定消息确认模式
  16.         channel.confirmSelect();
  18.         // 通过 channel 发送数据
  19.         // exchange：交换机，如果不传默认为 AMQP default
  20.         String confirmExchangeName = "confirm_exchange_name";
  21.         String confirmRoutingKey = "confirm_routingKey";
  22.         String confirmMsg = "confirm hello world";
  24.         channel.basicPublish(confirmExchangeName, confirmRoutingKey, null, confirmMsg.getBytes());
  26.         channel.addConfirmListener(new ConfirmListener() {
  27.             @Override
  28.             public void handleAck(long l, boolean b) throws IOException {
  29.                 System.out.println("--------handleAck-----------");
  30.                 System.out.print(l);
  31.                 System.out.print(b);
  32.             }
  34.             @Override
  35.             public void handleNack(long l, boolean b) throws IOException {
  36.                 System.out.println("--------handleNack-----------");
  37.                 System.out.print(l);
  38.                 System.out.print(b);
  39.             }
  40.         });
  42.         // 关闭链接
  43.         // channel.close();
  44.         // connection.close();
  45.     }
  46. }

源码地址

  1. https://github.com/Q-Angelo/SpringBoot-Course/tree/master/chapter8/chapter8-1/src/main/java/com/may/rabbitmq/confirm/helloworld