spring-boot + rabbitmq消息手动确认模式的几点说明(重试机制)


声明:本文转载自https://my.oschina.net/dengfuwei/blog/1595047,转载目的在于传递更多信息,仅供学习交流之用。如有侵权行为,请联系我,我会及时删除。

前提:使用rabbitmq的手动确认消息的模式

消息手动确认模式的几点说明

  • 监听的方法内部必须使用channel进行消息确认,包括消费成功或消费失败

  • 如果不手动确认,也不抛出异常,消息不会自动重新推送(包括其他消费者),因为对于rabbitmq来说始终没有接收到消息消费是否成功的确认,并且Channel是在消费端有缓存的,没有断开连接

  • 如果rabbitmq断开,连接后会自动重新推送(不管是网络问题还是宕机)

  • 如果消费端应用重启,消息会自动重新推送

  • 如果消费端处理消息的时候宕机,消息会自动推给其他的消费者

  • 如果监听消息的方法抛出异常,消息会按照listener.retry的配置进行重发,但是重发次数完了之后还抛出异常的话,消息不会重发(也不会重发到其他消费者),只有应用重启后会重新推送。因为retry是消费端内部处理的,包括异常也是内部处理,对于rabbitmq是不知道的(此场景解决方案后面有)

  • spring.rabbitmq.listener.retry配置的重发是在消费端应用内处理的,不是rabbitqq重发

  • 可以配置MessageRecoverer对异常消息进行处理,此处理会在listener.retry次数尝试完并还是抛出异常的情况下才会调用,默认有两个实现:

    • RepublishMessageRecoverer:将消息重新发送到指定队列,需手动配置,如:
    @Bean public MessageRecoverer messageRecoverer(RabbitTemplate rabbitTemplate){     return new RepublishMessageRecoverer(rabbitTemplate, "exchangemsxferror", "routingkeymsxferror"); } 
    • RejectAndDontRequeueRecoverer:如果不手动配置MessageRecoverer,会默认使用这个,实现仅仅是将异常打印抛出,源码如下:
    public class RejectAndDontRequeueRecoverer implements MessageRecoverer {      protected Log logger = LogFactory.getLog(RejectAndDontRequeueRecoverer.class);      @Override     public void recover(Message message, Throwable cause) {     	if (this.logger.isWarnEnabled()) {             this.logger.warn("Retries exhausted for message " + message, cause);     	}     	throw new ListenerExecutionFailedException("Retry Policy Exhausted", new AmqpRejectAndDontRequeueException(cause), message);     }  } 
  • 可以通过给队列(Queue)绑定死信队列,使用nack反馈给mq,会将消息转发到死信队列里面,此种方式需要自己在消费消息的方法内部将异常处理掉

//声明队列,并给队列增加x-dead-letter-exchange和x-dead-letter-routing-key参数,用于指定死信队列的路由和routingKey @Bean public Queue queue(){     Map<String, Object> args = new HashMap<String, Object>();     args.put("x-dead-letter-exchange",IntegralConstant.DEAD_EXCHANGE_NAME);     args.put("x-dead-letter-routing-key",IntegralConstant.DEAD_ROUTING_KEY);     return new Queue(IntegralConstant.QUEUE_NAME, true, false, false, args); }  //消息确认时使用nack,并且requeue参数传false channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false); 
  • 消息变成死信有以下几种情况:
    • 消息被拒绝(basic.reject/ basic.nack)并且requeue=false
    • 消息TTL过期(参考:RabbitMQ之TTL(Time-To-Live 过期时间))
    • 队列达到最大长度

总结

  • 消息监听内必须使用channel对消息进行确认,不管是确认消费成功还是确认消费失败
  • 消息监听内的异常处理有两种方式:
    • 内部catch后直接处理,然后使用channel对消息进行确认
    • 配置RepublishMessageRecoverer将处理异常的消息发送到指定队列专门处理或记录
  • 监听的方法内抛出异常貌似没有太大用处。因为抛出异常就算是重试也非常有可能会继续出现异常,当重试次数完了之后消息就只有重启应用才能接收到了,很有可能导致消息消费不及时。当然可以配置RepublishMessageRecoverer来解决,但是万一RepublishMessageRecoverer发送失败了呢。。那就可能造成消息消费不及时了。所以即使需要将处理出现异常的消息统一放到另外队列去处理,个人建议两种方式:
    • catch异常后,手动发送到指定队列,然后使用channel给rabbitmq确认消息已消费
    • 给Queue绑定死信队列,使用nack(requque为false)确认消息消费失败

本文发表于2017年12月26日 22:32
(c)注:本文转载自https://my.oschina.net/dengfuwei/blog/1595047,转载目的在于传递更多信息,并不代表本网赞同其观点和对其真实性负责。如有侵权行为,请联系我们,我们会及时删除.

阅读 3025 讨论 0 喜欢 0

抢先体验

扫码体验
趣味小程序
文字表情生成器

闪念胶囊

你要过得好哇,这样我才能恨你啊,你要是过得不好,我都不知道该恨你还是拥抱你啊。

直抵黄龙府,与诸君痛饮尔。

那时陪伴我的人啊,你们如今在何方。

不出意外的话,我们再也不会见了,祝你前程似锦。

这世界真好,吃野东西也要留出这条命来看看

快捷链接
网站地图
提交友链
Copyright © 2016 - 2021 Cion.
All Rights Reserved.
京ICP备2021004668号-1