Skip to content

面试速答(先看这里)

**一句话结论:**为了解决这个问题,有两种方案,一种是通过confirm机制,另外一种是事务机制,因为事务机制并不推荐,这里先介绍Confirm机制。

60秒标准回答:

当我们作为一个消息发送方,如何保证我们给RabbitMQ发送的消息一定能发送成功,如何确保他一定能收到这个消息呢?

我们知道,RabbitMQ的消息最终是存储在Queue上的,而在Queue之前还要经过Exchange,那么这个过程中就有两个地方可能导致消息丢失。第一个是Producer到Exchange的过程,第二个是Exchange到Queue的过程

为了解决这个问题,有两种方案,一种是通过confirm机制,另外一种是事务机制,因为事务机制并不推荐,这里先介绍Confirm机制

**答题顺序:**结论 → 原理/机制 → 关键流程 → 场景与取舍 → 易错点

回答主线:

  • **要点1:**当我们作为一个消息发送方,如何保证我们给RabbitMQ发送的消息一定能发送成功,如何确保他一定能收到这个消息呢?
  • **要点2:**我们知道,RabbitMQ的消息最终是存储在Queue上的,而在Queue之前还要经过Exchange,那么这个过程中就有两个地方可能导致消息丢失。
  • **要点3:**上面两个可能丢失的过程,都可以利用confirm机制,注册回调来监听是否成功。
  • **要点4:**Publisher Confirm 是一种机制,用于确保消息已经被Exchange成功接收和处理。
  • **要点5:**Publisher Returns 机制与Publisher Confirms类似,但用于处理在消息无法路由到任何队列时的情况。

**记忆锚点:**Exchange → RabbitMQ → Queue → Publisher → confirm → Confirm

易错提醒:

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

加分表达:

  • 一旦消息成功到达Exchange并被处理,RabbitMQ会向消息生产者发送确认信号(ACK)。
  • 如果由于某种原因(例如,Exchange不存在或路由键不匹配)消息无法被处理,RabbitMQ会向消息生产者发送否认信号(NACK)。
  • 当RabbitMQ在无法路由消息时将消息返回给消息生产者,但是如果能正确路由,则不会返回消息。

追问准备:

  • 围绕「Exchange」:底层原理是什么?使用时有哪些边界和常见坑?
  • 围绕「RabbitMQ」:底层原理是什么?使用时有哪些边界和常见坑?
  • 围绕「Queue」:底层原理是什么?使用时有哪些边界和常见坑?
  • 如果线上出现异常,你会如何定位、验证并规避?

典型回答 ​

当我们作为一个消息发送方,如何保证我们给RabbitMQ发送的消息一定能发送成功,如何确保他一定能收到这个消息呢?

我们知道,RabbitMQ的消息最终是存储在Queue上的,而在Queue之前还要经过Exchange,那么这个过程中就有两个地方可能导致消息丢失。第一个是Producer到Exchange的过程,第二个是Exchange到Queue的过程。

image.png

为了解决这个问题,有两种方案,一种是通过confirm机制,另外一种是事务机制,因为事务机制并不推荐,这里先介绍Confirm机制。

📄 介绍下RabbitMQ的事务机制

打开文档:介绍下RabbitMQ的事务机制

上面两个可能丢失的过程,都可以利用confirm机制,注册回调来监听是否成功。

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

plain
// 启用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在无法路由消息时将消息返回给消息生产者,但是如果能正确路由,则不会返回消息。

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

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

完整的代码如下:

plain
import com.rabbitmq.client.*;

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 replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("Message returned with replyCode: " + 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();
        }
    }
}

另外,这里如果发送到Queue之后,是否一定能持久化下来,是否一定不丢,这就是另外一个话题了。

📄 RabbitMQ如何保证消息不丢

打开文档:RabbitMQ如何保证消息不丢

​