如何保障消息一定能发送到RabbitMQ？

news2024/11/20 11:32:04

我们知道，RabbitMQ的消息最终是存储在Queue上的，而在Queue之前还要经过Exchange，那么这个过程中就有两个地方可能导致消息丢失。第一个是Producer到Exchange的过程，第二个是Exchange到Queue的过程。
在这里插入图片描述
为了解决这个问题，有两种方案，一种是通过confirm机制，另外一种是事务机制，因为事务机制并不推荐，这里先介绍Confirm机制。

Publisher Confirm是一种机制，用于确保消息已经被Exchange成功接收和处理。一旦消息成功到达Exchange并被处理，RabbitMQ会向消息生产者发送确认信号（ACK）。如果由于某种原因（例如，Exchange不存在或路由键不匹配）消息无法被处理，RabbitMQ会向消息生产者发送否定信号（NACK）。

//启用Publisher Confirms
            channel.confirmSelect();

            //设置Publisher Confirms回调
            channel.addConfirmListener(new ConfirmListener() {
                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    System.out.println("Message confirmed with deliveryTag:"+deliveryTag);
                    //在这里处理消息确认
                }

                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    System.out.println("Message not confirmed with deliveryTag:"+deliveryTag);
                    //在这里处理消息未确认
                }
            });

Publisher Returns机制与Publisher Confirms类似，但用于处理在消息无法路由到任何队列时的情况。当RabbitMQ在无法路由消息时将消息返回给消息生产者，但是如果能正常路由，则不会返回消息。

//启用Publisher Returns
            channel.addReturnListener(new ReturnListener() {
                @Override
                public void handleReturn(int replyCode, String replyTest, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("Message returned with replayCode: "+replyCode);
                    //在这里处理消息发送到Queue失败的返回
                }
            });

通过以上方式，我们注册了两个回调监听，用于在消息发送到Exchange或者Queue失败时进行异常处理。通常我们可以在失败时精心报警或者重试来保障一定能发送成功。

完整代码：

package com.example.demo.rabbitmq;

import com.rabbitmq.client.*;

import java.io.IOException;

public class PublisherCallbacksExample {
    public static void main(String[] args) throws Exception{
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("localhost");
        try(Connection connection=factory.newConnection();
            Channel channel=connection.createChannel()){

            //启用Publisher Confirms
            channel.confirmSelect();

            //设置Publisher Confirms回调
            channel.addConfirmListener(new ConfirmListener() {
                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    System.out.println("Message confirmed with deliveryTag:"+deliveryTag);
                    //在这里处理消息确认
                }

                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    System.out.println("Message not confirmed with deliveryTag:"+deliveryTag);
                    //在这里处理消息未确认
                }
            });

            //启用Publisher Returns
            channel.addReturnListener(new ReturnListener() {
                @Override
                public void handleReturn(int replyCode, String replyTest, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("Message returned with replayCode: "+replyCode);
                    //在这里处理消息发送到Queue失败的返回
                }
            });

            String exchangeName = "my_exchange";
            String routingKey = "my_routing_key";
            String message = "Hello,RabbitMQ!";

            //发布消息到Exchange
            channel.basicPublish(exchangeName,routingKey,true,null,message.getBytes());

            //等待Publisher Confirms
            if (!channel.waitForConfirms()) {
                System.out.println("Message was not confirmed!");
            }

            //关闭通道和连接
            channel.close();

        }

    }
}