RabbitMQ(五)消息发送失败后的处理

鸬鹚洲

　　就是消息发送成功或失败的时候,都会调用confirmListener 或者returnListener.
　　如果消息发送成功,就不考虑了.当消息发送失败时,怎么处理这个消息呢.
　　1.自动重发
　　2.系统预警人工处理等
　　以上操作,都需要知道是哪条消息,具体什么内容发送失败了,才能进行后续处理.
　　在returnListener中,参数是有消息内容,exchange,routingKey 这些内容的.
　　但是在confirmListener中,却是什么都没有,只有个correlationData,而我们打印出的correlationData都是null
　　看来问题的关键就在这个correlationData上了, CorrelationData类只有一个属性ID, 很明显,我们在发送消息时,将消息和correlationData的ID做一个绑定,就可以根据id拿到消息. 然后进行重发,报警等操作了.
　　到发送消息的类里面, 发现AmqpTemplate 里面竟然没有和CorrelationData相关的方法,没办法把CorrelationData.id和消息进行绑定..
　　原来需要使用RabbitTemplate 而不是 AmqpTemplate 类

　　找到方法后,下面就非常简单了.






[java] view plain copy


print?

• @Service("rabbitTemplatePublishService")  
  
• public class RabbitTemplatePublishService {  
  
•     @Autowired  
  
•     private RabbitTemplate rmqpTemplate;  
  
• 
  
•     public void send(String exchange, String routingKey, Object obj) {  
  
•         String msgId = UUID.randomUUID().toString();
  
•         Message message = MessageBuilder.withBody(obj.toString().getBytes())
  
•                 .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
  
•                 .setCorrelationId(msgId.getBytes()).build();
  
•         CorrelationData date = new CorrelationData(msgId);  
  
•         // TODO 将 msgId 与 message 的关系保存起来,例如放到缓存中  
  
•         rmqpTemplate.send(exchange, routingKey, message, date);
  
•     }
  
• }
  

@Service("rabbitTemplatePublishService")
public class RabbitTemplatePublishService {
@Autowired
private RabbitTemplate rmqpTemplate;
public void send(String exchange, String routingKey, Object obj) {
String msgId = UUID.randomUUID().toString();
Message message = MessageBuilder.withBody(obj.toString().getBytes())
.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
.setCorrelationId(msgId.getBytes()).build();
CorrelationData date = new CorrelationData(msgId);
// TODO 将 msgId 与 message 的关系保存起来,例如放到缓存中
rmqpTemplate.send(exchange, routingKey, message, date);
}
}
　　


　　上面的代码,message中,也可以setCorrelationId为自己指定的id,这样方便confirm和return一起处理.
　　
上面将msgId和message关系保存后, 在confirmListener或者returnListener中,就可以根据msgId获取message进行对于操作.
　　上面的思路看起来解决了失败后消息处理的问题.但是实际上却是不可靠的
　　3.如果进行confirm的时候,或者return的时候, 失败了怎么办?  
　　4.msgId和message的关系要保存多久?
　　但是上面的思路已经抓到了问题的重点,即msgId和message做绑定.
　　针对3,4两个问题,只需要变通下即可解决
　　方案如下:
　　发送消息前,绑定并保存msgId和message的关系
　　当confirm或return回调时,根据ack类别等,分别处理. 例如return或者ack=false则说明有问题,报警, 如果ack=true则删除关系
　　(因为return在confirm前,所以一条消息在return后又ack=true的情况也是按return处理)
　　定时检查这个绑定关系列表,如果发现一些已经超时(自己设定的超时时间)未被处理(即未return和confirm),则手动处理这些消息.
　　至此,发送消息基本已经没问题了.
　　需要注意如果是自动重发的话,消费端需要做幂等或去重处理.