RabbitMQ消息的应答

简介: 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));
            }
        });
    }
}

效果展示

此时把消费者1停掉,那么上面这10条为确认的消费会重新入队,发送给另外的消费者。


相关实践学习
RocketMQ一站式入门使用
从源码编译、部署broker、部署namesrv,使用java客户端首发消息等一站式入门RocketMQ。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
3天前
|
消息中间件 存储 缓存
RabbitMQ之消息应答和持久化
【1月更文挑战第11天】 一、消息应答 1.概念 2.自动应答 3.消息应答方法 4.Multiple 的解释 5.消息自动重新入队 6.消息手动应答代码 7.手动应答效果演示 二、RabbitMQ持久化 1.概念 2.队列如何实现持久化 3.消息实现持久化 4.不公平分发 5.预取值
199 2
|
6月前
|
消息中间件
消息中间件系列教程(16) -RabbitMQ-应答模式
消息中间件系列教程(16) -RabbitMQ-应答模式
40 0
|
9月前
|
消息中间件 存储 缓存
RabbitMQ (HelloWord 消息应答 持久化 不公平分发 预取值)2
RabbitMQ (HelloWord 消息应答 持久化 不公平分发 预取值)2
44 0
|
9月前
|
消息中间件
RabbitMQ (HelloWord 消息应答 持久化 不公平分发 预取值)1
RabbitMQ (HelloWord 消息应答 持久化 不公平分发 预取值)1
44 0
|
消息中间件 网络协议
RabbitMQ学习(四):消息应答
为了保证消息在发送过程中不丢失,rabbitmq 引入消息应答机制,消息应答就是:消费者在接 收到消息并且处理该消息之后,告诉 rabbitmq 它已经处理了,rabbitmq 可以把该消息删除了。
144 0
RabbitMQ学习(四):消息应答
|
消息中间件 网络协议
RabbitMQ消息应答
消费者完成一个任务可能需要一段时间,如果其中一个消费者处理一个长的任务并仅只完成了部分突然它挂掉了,会发生什么情况。RabbitMQ一旦向消费者传递了一条消息,便立即将该消息标记为删除。
RabbitMQ消息应答
|
消息中间件 网络协议 Java
RabbitMQ——消息手动应答、队列/消息持久化、不公平分发、预取值的概念理解及应用举例
RabbitMQ——消息手动应答、队列/消息持久化、不公平分发、预取值的概念理解及应用举例
RabbitMQ——消息手动应答、队列/消息持久化、不公平分发、预取值的概念理解及应用举例
|
消息中间件
Rabbitmq手动应答代码实现与测试
rabbitmq作为现在的主流消息队列软件,有着他自己独特的功能与特性,今天来说一下其中的应答机制。
145 0
Rabbitmq手动应答代码实现与测试
|
消息中间件
轻松搞定RabbitMQ3:消息应答与消息持久化
轻松搞定RabbitMQ(三)——消息应答与消息持久化 这个官网的第二个例子中的消息应答和消息持久化部分。我把它摘出来作为单独的一块儿来分享。
|
消息中间件
轻松搞定RabbitMQ(三)——消息应答与消息持久化
       这个官网的第二个例子中的消息应答和消息持久化部分。我把它摘出来作为单独的一块儿来分享。 Message acknowledgment(消息应答)        执行一个任务可能需要花费几秒钟,你可能会担心如果一个消费者在执行任务过程中挂掉了。
1231 0