RabbitMQ 可靠投递
标签: RabbitMQ shovel-plugin ConfirmCallback RabbitMQ消息投递
背景
confirmCallback 确认模式
returnCallback 未投递到 queue 退回模式
shovel-plugin 跨机房可靠投递
背景在使用 RabbitMQ 的时候,作为消息发送方希望杜绝任何消息丢失或者投递失败场景。RabbitMQ 为我们提供了两个选项用来控制消息的投递可靠性模式。
rabbitmq 整个消息投递的路径为:
producer->rabbitmq broker cluster->exchange->queue->consumer
message 从 producer 到 rabbitmq broker cluster 则会返回一个 confirmCallback 。
message 从 exchange->queue 投递失败则会返回一个 returnCallback 。我们将利用这两个 callback 控制消息的最终一致性和部分纠错能力。
在创建 connectionFactory 的时候设置 PublisherConfirms(true) 选项,开启 confirmcallback 。
CachingConnectionFactory factory = new CachingConnectionFactory(); factory.setPublisherConfirms(true);//开启confirm模式 RabbitTemplate rabbitTemplate = new RabbitTemplate(factory); rabbitTemplate.setConfirmCallback((data, ack, cause) -> { if (!ack) { log.error("消息发送失败!" + cause + data.toString()); } else { log.info("消息发送成功,消息ID:" + (data != null ? data.getId() : null)); } });我们来看下 ConfirmCallback 接口。
public interface ConfirmCallback { /** * Confirmation callback. * @param correlationData correlation data for the callback. * @param ack true for ack, false for nack * @param cause An optional cause, for nack, when available, otherwise null. */ void confirm(CorrelationData correlationData, boolean ack, String cause); }重点是 CorrelationData 对象,每个发送的消息都需要配备一个 CorrelationData 相关数据对象,CorrelationData 对象内部只有一个 id 属性,用来表示当前消息唯一性。
发送的时候创建一个 CorrelationData 对象。
User user = new User(); user.setID(1010101L); user.setUserName("plen"); rabbitTemplate.convertAndSend(exchange, routing, user, message -> { message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT); return message; }, new CorrelationData(user.getID().toString()));这里将 user ID 设置为当前消息 CorrelationData id 。当然这里是纯粹 demo,真实场景是需要做业务无关消息 ID 生成,同时要记录下这个 id 用来纠错和对账。
消息只要被 rabbitmq broker 接收到就会执行 confirmCallback,如果是 cluster 模式,需要所有 broker 接收到才会调用 confirmCallback。
被 broker 接收到只能表示 message 已经到达服务器,并不能保证消息一定会被投递到目标 queue 里。所以需要用到接下来的 returnCallback 。
returnCallback 未投递到queue退回模式confrim 模式只能保证消息到达 broker,不能保证消息准确投递到目标 queue 里。在有些业务场景下,我们需要保证消息一定要投递到目标 queue 里,此时就需要用到 return 退回模式。
同样创建 ConnectionFactory 到时候需要设置 PublisherReturns(true) 选项。
CachingConnectionFactory factory = new CachingConnectionFactory(); factory.setPublisherReturns(true);//开启return模式 rabbitTemplate.setMandatory(true);//开启强制委托模式 rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> log.info(MessageFormat.format("消息发送ReturnCallback:{0},{1},{2},{3},{4},{5}", message, replyCode, replyText, exchange, routingKey)));这样如果未能投递到目标 queue 里将调用 returnCallback ,可以记录下详细到投递数据,定期的巡检或者自动纠错都需要这些数据。
shovel-plugin 跨机房可靠投递RabbitMQ 在跨机房集成提供了一个不错的插件 shovel 。使用 shovel-plugin 插件非常方便,shovel 可以接受机房之间的网络断开、机器下线等不稳定因素。
这里有两个 broker :
10.211.55.3 rabbit_node1
10.211.55.4 rabbit_node2
我们希望将发送给 rabbit_node1 plen.queue 的消息传输到 rabbit_node2 plen.queue 中。我们先开启 rabbit_node1 的 shovel-plugin。