RabbitMQ发送方确认机制

2023-05-23 83

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

本文涉及的产品

容器服务 Serverless 版 ACK Serverless，952元额度多规格

容器服务 Serverless 版 ACK Serverless，317元额度多规格

简介： RabbitMQ发送方确认机制

1、前言

RabbitMQ消息首先发送到交换机，然后通过路由键【routingKey】和【bindingKey】比较从而将消息发送到对应的队列【queue】上。在这个过程有两个地方消息可能会丢失：

消息发送到交换机的过程。
消息从交换机发送到队列的过程。

而RabbitMQ提供了类似于回调函数的机制来告诉发送方消息是否发送成功。这里针对上述的两种情况，RabbitMQ也是给出了以下的应对策略：

publisher-confirm：消息到达交换机时会触发。
publisher-return：到达交换机但是没有路由到队列，会返回ack以及失败原因。

2、publisher-confirm

在SpringBoot项目的properties文件中加上

spring.rabbitmq.publisher-confirm-type=correlated

该配置有三个值：

none：是禁用发布确认模式，是默认值
correlated：是发布消息成功到交换器后会触发回调方法
simple：有两种效果，第一种和correlated值一样会触发回调方法；第二种在发布消息成功后使用rabbitTemplate调用waitForConfirms或waitForConfirmsOrDie方法等待broker节点返回发送结果，根据返回结果来判定下一步的逻辑，要注意的点是waitForConfirmsOrDie方法如果返回false则会关闭channel，则接下来无法发送消息到broker。

RabbitMQ的配置类实现ConfirmCallback

/*** @author LoneWalker* @date 2023/4/8* @description*/@Slf4j@ConfigurationpublicclassRabbitMqConfigimplementsRabbitTemplate.ConfirmCallback {
@BeanpublicRabbitTemplaterabbitTemplate(CachingConnectionFactoryconnectionFactory) {
RabbitTemplaterabbitTemplate=newRabbitTemplate(connectionFactory);
rabbitTemplate.setMessageConverter(jackson2JsonMessageConverter());
//设置给rabbitTemplaterabbitTemplate.setConfirmCallback(this);
returnrabbitTemplate;
    }
@BeanpublicMessageConverterjackson2JsonMessageConverter() {
returnnewJackson2JsonMessageConverter();
    }
@BeanpublicDirectExchangegetExchange(){
returnnewDirectExchange("directExchange",false,false);
    }
@BeanpublicQueuegetQueue(){
returnnewQueue("publisher.addUser",true,false,false);
    }
@BeanpublicBindinggetBinding(DirectExchangeexchange,Queuequeue){
returnBindingBuilder.bind(queue).to(exchange).with("publisher.addUser");
    }
/*** 消息成功到达交换机会触发* @param correlationData* @param ack* @param cause*/@Overridepublicvoidconfirm(CorrelationDatacorrelationData, booleanack, Stringcause) {
if (ack) {
log.info("交换机收到消息成功："+correlationData.getId());
        }else {
log.error("交换机收到消息失败："+correlationData.getId() +"原因："+cause);
        }
    }
}

而需要这个correlationData是因为确认机制发送消息时,需要给每个消息设置一个全局唯一id,以区分不同消息,避免ack冲突。所以我们改写一下发送消息的方法：

@RequiredArgsConstructor@ServicepublicclassPublisherServiceImplimplementsPublisherService{
privatefinalRabbitTemplaterabbitTemplate;
@OverridepublicvoidaddUser(Useruser) {
CorrelationDatacorrelationData=newCorrelationData();
correlationData.setId(UUID.randomUUID().toString());
rabbitTemplate.convertAndSend("directExchange","publisher.addUser",user,correlationData);
    }
}

然后发送消息：

再模拟一下失败的情况——把交换机名称改成错的：

温馨提示：测试完把交换机名称改回去。

3、publisher-return

在SpringBoot项目的properties文件中添加：

spring.rabbitmq.publisher-returns=true

###消息在没有被队列接收时是否强行退回还是直接丢弃

spring.rabbitmq.template.mandatory=true

RabbitMQ的配置类再实现ReturnsCallback

@Slf4j@ConfigurationpublicclassRabbitMqConfigimplementsRabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback {
@BeanpublicRabbitTemplaterabbitTemplate(CachingConnectionFactoryconnectionFactory) {
RabbitTemplaterabbitTemplate=newRabbitTemplate(connectionFactory);
rabbitTemplate.setMessageConverter(jackson2JsonMessageConverter());
//设置给rabbitTemplaterabbitTemplate.setConfirmCallback(this);
rabbitTemplate.setReturnsCallback(this);
rabbitTemplate.setMandatory(true);
returnrabbitTemplate;
    }
@BeanpublicMessageConverterjackson2JsonMessageConverter() {
returnnewJackson2JsonMessageConverter();
    }
@BeanpublicDirectExchangegetExchange(){
returnnewDirectExchange("directExchange",false,false);
    }
@BeanpublicQueuegetQueue(){
returnnewQueue("publisher.addUser",true,false,false);
    }
@BeanpublicBindinggetBinding(DirectExchangeexchange,Queuequeue){
returnBindingBuilder.bind(queue).to(exchange).with("publisher.addUser");
    }
/*** 消息成功到达交换机会触发* @param correlationData* @param ack* @param cause*/@Overridepublicvoidconfirm(CorrelationDatacorrelationData, booleanack, Stringcause) {
if (ack) {
log.info("交换机收到消息成功："+correlationData.getId());
        }else {
log.error("交换机收到消息失败："+correlationData.getId() +"原因："+cause);
        }
    }
/*** 消息未成功到达队列会触发* @param returnedMessage*/@OverridepublicvoidreturnedMessage(ReturnedMessagereturnedMessage) {
log.error("{}--消息未成功到达队列",returnedMessage.getMessage().getMessageProperties().getMessageId());
    }
}

把路由键改为错误的值：

正常来说消息到达交换机就一定可以到达队列，到不了队列基本上就是代码写错了。

RabbitMQ发送方确认机制

1、前言

2、publisher-confirm

3、publisher-return

热门文章

最新文章

相关课程

相关电子书

相关实验场景