rabbitmq消息接收的可靠性机制

消费端的两种处理机制:

两种机制的区别, 第一种是在消费端出现异常, 系统执行的, 如果多次重试失败, 则可以抛出指定异常拒绝该消息(等同与reject)或者将消息发送到指定队列;

第二种ack机制必须要内部catch住消费者的异常, 手动的进行ack或者nack给rabbitmq , 然后rabbitmq根据配置重新发送消息或者直接舍弃该消息

1. spring.rabbitmq.listener.retry配置的重发是在消费端应用内处理的，不是rabbitqq重发. 因为retry是消费端内部处理的，包括异常也是内部处理，对于rabbitmq是不知道的

2.消息ack

a. 返回ack则表明消息被处理, rabbitmq删除消息,

b. nack或者reject并且requeue=true则rabbitmq重新发送该消息,requeue=false则讲消息丢弃或放入私信队列

    nack和reject区别, 前者可以批量拒绝 消息, reject只能单条消息拒绝

c. 如果rabbitmq发出的消息一直得不到服务端的相应, 则在

         rabbitmq断开，连接后会自动重新推送（不管是网络问题还是宕机）

         消费端应用重启，消息会自动重新推送

关于retry

spring.rabbitmq.listener.simple.retry.enabled=true
spring.rabbitmq.listener.simple.retry.max-attempts=3
spring.rabbitmq.listener.simple.acknowledge-mode=manual

@RabbitListener(queues = "queue.item.search.test")public void DocTest(org.springframework.amqp.core.Message message, Channel channel) {System.out.println("****************************");System.out.println(message.getBody());if (true)throw new YuncaiException("");try {channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);} catch (IOException e) {e.printStackTrace();}}

rabbitmq只做一次消息发送, 消费者内部自动重发了两次(1+2=3次), 之后消息停止发送, rabbitmq的web控制台unack数为1

关于springamqp做的autoAck(默认)

在当前线程结束没有抛出异常的情况下才会ack, 抛出异常则执行retry机制

总结:

如果使用springamqp默认的autoack,并且配置类retry次数, 则在抛出异常重试指定次数后消息被丢弃,相当于消费端发送了ack信号;

如果使用manualack,并且配置了retry次数, 在ack前抛出异常, 则在重试指定次数后, 该消息在broker中还是unack状态;

如果使用manualack,不管配置配置了retry与否, 包住所有可能的异常, 手动ack或者nack,决定broker是ack还是unack;

以上三种情况在接受消息的过程中, 消费者和broker连接中断((消费者宕掉),消息都会重新发送

本文来自互联网用户投稿，文章观点仅代表作者本人，不代表本站立场，不承担相关法律责任。如若转载，请注明出处。 如若内容造成侵权/违法违规/事实不符，请点击【内容举报】进行投诉反馈！