RabbitMQ消息的应答-阿里云开发者社区

RabbitMQ消息的应答

2024-04-12 26

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： RabbitMQ消息的应答

消息的应答机制

消费者完成一个任务可能需要一段时间，如果其中一个消费者处理一个长的任务并仅只完成了部分突然它挂掉了，会发生什么情况。RabbitMQ 一旦向消费者传递了一条消息，便立即将该消息标记为删除。在这种情况下，突然有个消费者挂掉了，我们将丢失正在处理的消息。以及后续发送给该消费者的消息，因为它无法接收到。

为了保证消息在发送过程中不丢失，RabbitMQ 引入消息应答机制，消息应答就是：消费者在接收到消息并且处理该消息之后，告诉 RabbitMQ 它已经处理了， RabbitMQ 可以把该消息删除了。

自动应答

消息发送后立即被认为已经传送成功，这种模式需要在高吞吐量和数据传输安全性方面做权衡，因为这种模式如果消息在接收到之前，消费者那边出现连接或者 channel 关闭，那么消息就丢失了，当然另一方面这种模式消费者那边可以传递过载的消息，没有对传递的消息数量进行限制，当然这样有可能使得消费者这边由于接收太多还来不及处理的消息，导致这些消息的积压，最终使得内存耗尽，最终这些消费者线程被操作系统杀死，所以这种模式仅适用在消费者可以高效并以某种速率能够处理这些消息的情况下使用。

手动应答

手动应答的好处是可以批量应答并且减少网络拥堵，批量应答的批量范围是channel 上未应答的消息。比如说 channel 上有传送 tag 的消息5，6，7，8 当前 tag 是8 那么此时5-8 的这些还未应答的消息都会被确认收到消息应答。但是实际上还是不建议开启批量应答的。

Channel.basicAck(用于肯定确认)

RabbitMQ 已知道该消息并且成功的处理消息，可以将其丢弃了。

Channel.basicNack(用于否定确认)

不处理该消息了直接拒绝，可以将其丢弃了。

Channel.basicReject(用于否定确认)

与 Channel.basicNack 相比少一个参数。

不处理该消息了直接拒绝，可以将其丢弃了。

消息自动重新入队

如果消费者由于某些原因失去连接(其通道已关闭，连接已关闭或 TCP 连接丢失)，导致消息未发送 ACK 确认，RabbitMQ 将了解到消息未完全处理，并将对其重新排队。如果此时其他消费者可以处理，它将很快将其重新分发给另一个消费者。这样，即使某个消费者偶尔死亡，也可以确保不会丢失任何消息。

代码实现

生产者

public class MyProducer {
    @Test
    public void test() throws Exception {
        // 队列名称
        String queue = "xw_queue";
        String message = "Hello World -> ";
        // 创建工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setVirtualHost("/");
        factory.setHost("xuewei.world");
        factory.setUsername("xuewei");
        factory.setPassword("123456");
        factory.setPort(5672);
        // 创建连接和通道
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        for (int i = 0; i < 20; i++) {
            // 发布消息
            channel.basicPublish("xw_exchange", queue, null, (message + i).getBytes());
        }
    }
}

消费者1

开启手动确认后，消费者1如果在处理消息的回调中不确认消息，那么队列中的消息会处于unacked的状态，如果消费者1突然挂掉，那么这些未确认的消费会重新发送给其他消费者。

public class MyConsumer1 {
    public static void main(String[] args) throws Exception {
        // 队列名称
        String queue = "xw_queue";
        // 创建工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setVirtualHost("/");
        factory.setHost("xuewei.world");
        factory.setUsername("xuewei");
        factory.setPassword("123456");
        factory.setPort(5672);
        // 创建连接和通道
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(queue, true, false, false, null);
        channel.queueBind("", "xw_exchange", queue);
        // 配置开启手动应答
        channel.basicConsume(queue, false, new DefaultConsumer(channel) {
            @SneakyThrows
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                // 消费完成后手动应答
                // channel.basicAck(envelope.getDeliveryTag(), false);
                Thread.sleep(5000);
                System.out.println("消费者1：接收到消息: " + new String(body));
            }
        });
    }
}

消费者2

消费者2正常消费消息，收到消息后立刻确认。

public class MyConsumer2 {
    public static void main(String[] args) throws Exception {
        // 队列名称
        String queue = "xw_queue";
        // 创建工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setVirtualHost("/");
        factory.setHost("xuewei.world");
        factory.setUsername("xuewei");
        factory.setPassword("123456");
        factory.setPort(5672);
        // 创建连接和通道
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(queue, true, false, false, null);
        channel.queueBind("", "xw_exchange", queue);
        // 配置开启手动应答
        channel.basicConsume(queue, false, new DefaultConsumer(channel) {
            @SneakyThrows
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                // 消费完成后手动应答
                channel.basicAck(envelope.getDeliveryTag(), false);
                System.out.println("消费者2：接收到消息: " + new String(body));
            }
        });
    }
}