rabbitmq生产者确认模式------------------发送方确认模式

刺骨的言语ヽ痛彻心扉 2022-12-22 06:16 195阅读 0赞

**1、原理：生产者将信道设置成 confirm 模式，一旦信道进入 confirm 模式，所有在该信道上面发布的消息都将会被指派一个唯一的 ID(从 1 开始)，由这  
个 id 在生产者和 RabbitMQ 之间进行消息的确认。**

// 启用发送者确认模式
            channel.confirmSelect();

*a、不可路由的消息，当交换器发现，消息不能路由到任何队列，会进行确认操作，表示收到了消息。如果发送方设置了 mandatory 模式,则会先调用  
addReturnListener 监听器。*

*监听器的代码：*

// 添加失败者通知
            channel.addReturnListener(new ReturnListener() {
                public void handleReturn(int replyCode, String replyText,
                                         String exchange, String routingKey,
                                         AMQP.BasicProperties properties,
                                         byte[] body)
                        throws IOException {
                    String message = new String(body);
                    System.out.println("RabbitMq路由失败:  "+routingKey+"."+message);
                }
            });

*b、可路由的消息，要等到消息被投递到所有匹配的队列之后，broker 会发送一个确认给生产者(包含消息的唯一 ID)，这就使得生产者知道消息已经正确  
到达目的队列了，如果消息和队列是可持久化的，那么确认消息会在将消息写入磁盘之后发出，broker 回传给生产者的确认消息中 delivery-tag 域包含了  
确认消息的序列号*

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70][]

confirm 模式最大的好处在于他可以是异步的，一旦发布一条消息，生产者应用程序就可以在等信道返回确认的同时继续发送下一条消息，当消息最  
终得到确认之后，生产者应用便可以通过回调方法来处理该确认消息，如果 RabbitMQ 因为自身内部错误导致消息丢失，就会发送一条 nack 消息，生产  
者应用程序同样可以在回调方法中处理该 nack 消息决定下一步的处理。

**Confirm  的三种实现方式：**  
**方式一：**channel.waitForConfirms()普通发送方确认模式；消息到达交换器，就会返回 true。---------同步方法

//所有日志严重性级别
            for(int i=0;i<2;i++){
                // 发送的消息
                String message = "Hello World_"+(i+1);
                //参数1：exchange name
                //参数2：routing key
                channel.basicPublish(EXCHANGE_NAME, ROUTE_KEY, true,null, message.getBytes());
                System.out.println(" Sent Message: [" + ROUTE_KEY +"]:'"+ message + "'");
                //TODO
                //确认是否成功(true成功)
                if(channel.waitForConfirms()){
                    //发送成功
                    System.out.println("send success");                
                }else{
                    //发送失败，在这里决定发送失败的消息的业务逻辑
                    System.out.println("send failure");
                }
            }

**方式二：**channel.waitForConfirmsOrDie()批量确认模式；使用同步方式等所有的消息发送之后才会执行后面代码，只要有一个消息未到达交换器就会  
抛出 IOException 异常。---------同步方法

//TODO
            // 启用发送者确认模式（批量确认）
            channel.waitForConfirmsOrDie();

**方式三：**channel.addConfirmListener()异步监听发送方确认模式；---------------------异步方法

// 添加发送者确认监听器
            channel.addConfirmListener(new ConfirmListener() {
                //TODO 成功
                public void handleAck(long deliveryTag, boolean multiple)
                        throws IOException {
                    //发送成功，可以进行一些日志打印
                    System.out.println("send_ACK:"+deliveryTag+",multiple:"+multiple);
                }
                //TODO 失败
                public void handleNack(long deliveryTag, boolean multiple)
                        throws IOException {
                    //发送失败，可以进行失败的消息的业务处理
                    System.out.println("Erro----send_NACK:"+deliveryTag+",multiple:"+multiple);
                }
            });

**备用交换器**

在第一次声明交换器时被指定，用来提供一种预先存在的交换器，如果主交换器无法路由消息，那么消息将被路由到这个新的备用交换器。

/TODO
            // 声明备用交换器
            Map<String,Object> argsMap = new HashMap<String,Object>();
            argsMap.put("alternate-exchange",BAK_EXCHANGE_NAME);
            //主交换器
            channel.exchangeDeclare(EXCHANGE_NAME,"direct",
                    false,false,argsMap);
            //备用交换器
            channel.exchangeDeclare(BAK_EXCHANGE_NAME,BuiltinExchangeType.FANOUT,
                    true,false,null);

如果发布消息时同时设置了 mandatory 会发生什么？如果主交换器无法路由消息，RabbitMQ 并不会通知发布者，因为，向备用交换器发送消息，表  
示消息已经被路由了。注意，新的备用交换器就是普通的交换器，没有任何特殊的地方。  
使用备用交换器，向往常一样，声明 Queue 和备用交换器，把 Queue 绑定到备用交换器上。然后在声明主交换器时，通过交换器的参数，  
alternate-exchange,，将备用交换器设置给主交换器。  
建议备用交换器设置为 faout 类型，Queue 绑定时的路由键设置为“\#”

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70]: /images/20221120/48da558cdf6d4bde956fa8e4cdfb5783.png