如何保证RabbitMQ的可靠性传输

文章目录

producer到broke

发送方确认

生产者到交换机：confirm

可能因为网络的原因

交换机到队列：returns模式

代码或者配置层错误，导致路由是失败
（快递地址写错了）
队列也是有一些特性：长度，有效期

队列溢出：可以采用死信等方式

消息对列自身数据丢失（持久化）

方法名方法功能所属接口接口所属类confirm()确认消息是否发送到交换机ConfirmCallbackRabbitTemplatereturnedMessage()确认消息是否发送到队列ReturnsCallbackRabbitTemplate

①ConfirmCallback接口

这是RabbitTemplate内部的一个接口，源代码如下：

/**
     * A callback for publisher confirmations.
     *
     */@FunctionalInterfacepublicinterfaceConfirmCallback{/**
         * 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.
         */voidconfirm(@NullableCorrelationData correlationData,boolean ack,@NullableString cause);}

生产者端发送消息之后，回调confirm()方法

• ack参数值为true：表示消息成功发送到了交换机
• ack参数值为false：表示消息没有发送到交换机

②ReturnCallback接口

同样也RabbitTemplate内部的一个接口，源代码如下：

/**
     * A callback for returned messages.
     *
     * @since 2.3
     */@FunctionalInterfacepublicinterfaceReturnsCallback{/**
         * Returned message callback.
         * @param returned the returned message and metadata.
         */voidreturnedMessage(ReturnedMessage returned);}

注意：接口中的returnedMessage()方法仅在消息没有发送到队列时调用

ReturnedMessage类中主要属性含义如下：
属性名类型含义messageorg.springframework.amqp.core.Message消息以及消息相关数据replyCodeint应答码，类似于HTTP响应状态码replyTextString应答码说明exchangeString交换机名称routingKeyString路由键名称

packagecom.atguigu.mq.config;importjakarta.annotation.PostConstruct;importlombok.extern.slf4j.Slf4j;importorg.springframework.amqp.core.ReturnedMessage;importorg.springframework.amqp.rabbit.connection.CorrelationData;importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.stereotype.Component;@Component@Slf4jpublicclassMQProducerAckConfigimplementsRabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnsCallback{@AutowiredprivateRabbitTemplate rabbitTemplate;@PostConstructpublicvoidinit(){
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnsCallback(this);}@Overridepublicvoidconfirm(CorrelationData correlationData,boolean ack,String cause){if(ack){
            log.info("消息发送到交换机成功！数据："+ correlationData);}else{
            log.info("消息发送到交换机失败！数据："+ correlationData +" 原因："+ cause);}}@OverridepublicvoidreturnedMessage(ReturnedMessage returned){
        log.info("消息主体: "+newString(returned.getMessage().getBody()));
        log.info("应答码: "+ returned.getReplyCode());
        log.info("描述："+ returned.getReplyText());
        log.info("消息使用的交换器 exchange : "+ returned.getExchange());
        log.info("消息使用的路由键 routing : "+ returned.getRoutingKey());}}

Broke内部

持久化
交换机，队列，消息

在后台管理界面创建交换机和队列时，默认就是持久化的模式。
此时消息也是持久化的，不需要额外设置。

创建备份交换机
备份交换机一定要选择fanout类型，因为原交换机转入备份交换机时并不会指定路由键

Broke到达消费者

消费者异常，消息还没到消费者，比如说消费者宕机，消费逻辑有问题

手动确认

默认情况下，消费端取回消息后，默认会自动返回ACK确认消息，所以在前面的测试中消息被消费端消费之后，RabbitMQ得到ACK确认信息就会删除消息

但实际开发中，消费端根据消息队列投递的消息执行对应的业务，未必都能执行成功，如果希望能够多次重试，那么默认设定就不满足要求了

(但是可靠性不是越高越好，可靠性高性能有可能降低！)

需要根据业务需求