关于RabbitMq 生产者消息丢失问题

拼搏现实的明天。 2024-04-17 06:26 97阅读 0赞

rabbitmq作为优秀的消息队列中间件，估计大家都会用到。但是在实际过程中，生产者会存在消息丢失的情况。  
如下示例，总共发送了30W条消息，队列里却只有299954条信息，丢失了46条，对于精度要求很高的应用，这是不可接受的：

logger.info("start");
            for (int i = 0; i < 300000; i++) {
                RabbitMqRunner runner = new RabbitMqRunner(rabbitUtil, exchangeName, routingKeyName,
                        "这是第 " + i + " 个信息");
                taskExecutor.execute(runner);
            }
            logger.info("end");

![在这里插入图片描述][20190826201443471.png]

## 1. 配置缓存模式，解决因connection closed错误导致丢数据 ##

缓存模式一般两种：  
void setCacheMode(CachingConnectionFactory.CacheMode cacheMode)  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rhc2h1YWlnZWdlNjQy_size_16_color_FFFFFF_t_70] **CHANNEL模式**  
默认是该模式。程序运行期间ConnectionFactory只维护着一个connection，但是可以含有多个channel，操作rabbitmq之前必须先获取一个channel，否则将会阻塞。  
相关参数配置：  
**connectionFactory.setChannelCacheSize(100);**  
设置每个Connection中的缓存Channel的数量，默认值为25。此处配置相当于AMQP线程池，操作rabbitmq之前（send/receive message等）要先获取到一个Channel，获取Channel时会先从缓存中找闲置的Channel，如果没有则创建新的Channel，当Channel数量大于缓存数量时，多出来没法放进缓存的会被关闭。  
**connectionFactory.setChannelCheckoutTimeout(600);**  
单位毫秒，当这个值大于0时，ChannelCacheSize代表的是缓存的数量上限，当缓存获取不到可用的channel时，不会创建新的channel会等待指定的时间，若到时间后还获取不到可用的channel，直接抛出AmqpTimeoutException。  
若是以上配置没有设置，则可以在rabbitmq admin控制台中观察到channels不停刷新，关开关开，浪费性能。  
注意：在CONNECTION模式，这个值也会影响获取Connection的等待时间，超时获取不到Connection也会抛出AmqpTimeoutException异常。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rhc2h1YWlnZWdlNjQy_size_16_color_FFFFFF_t_70 1]  
**CONNECTION模式**  
CONNECTION模式。在这个模式下允许创建多个connection，会缓存一定数量的connection，每个connection中同样缓存着一些channel。  
相关参数配置：  
**connectionFactory.setConnectionCacheSize(3);**  
仅在CONNECTION模式下使用，指定connection缓存数量。  
**connectionFactory.setConnectionLimit(10);**  
仅在CONNECTION模式下使用，指定connection数量上限。

## 2.开始事务模式或者confirm模式 ##

\*\*开启事务会大幅降低消息发送及接收效率，因为当已经有一个事务存在时，后面的消息是不能被发送或者接收（对同一个consumer而言）的。\*\*因此此处不考虑事务模式，直接选择confirm模式。

//confirmCallBack：消息从生产者到达exchange时返回ack，消息未到达exchange返回nack
            connectionFactory.setPublisherConfirms(true);
            //returnCallBack：消息进入exchange但未进入queue时会被调用。
            connectionFactory.setPublisherReturns(true);
    
            //生产者确认
            rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
                @Override
                public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                    if (ack) {
                        logger.info("消息发送成功:correlationData({}),ack({}),cause({})", correlationData, ack, cause);
                    }
    
                }
            });
            rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
                @Override
                public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                    logger.warn("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}", exchange, routingKey, replyCode, replyText, message);
                }
            });

还有一个参数需要说下：mandatory。这个参数为true表示如果发送消息到了RabbitMq，没有对应该消息的队列。那么会将消息返回给生产者，此时仍然会发送ack确认消息。  
在实际运行过程中，会发现输出如下：  
![在这里插入图片描述][20190827105336985.png]  
correlationData的值始终为null。查看源码：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rhc2h1YWlnZWdlNjQy_size_16_color_FFFFFF_t_70 2]  
到了这里就明了了，我们需要设置Id和message；

/**
         * 指定 exchangeName  routingKeyName，发送payload
         *
         * @param exchangeName
         * @param routingKeyName
         * @param payload
         * @throws JsonProcessingException
         */
        public void convertAndSend(String exchangeName, String routingKeyName, Object payload) throws JsonProcessingException {
            try {
                rabbitTemplate.setExchange(exchangeName);
                rabbitTemplate.setRoutingKey(routingKeyName);
    
    
                Message message = MessageBuilder.withBody(objectMapper.writeValueAsBytes(payload)).setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();
                message.getMessageProperties().setHeader(AbstractJavaTypeMapper.DEFAULT_CONTENT_CLASSID_FIELD_NAME, MessageProperties.CONTENT_TYPE_JSON);
                message.getMessageProperties().setContentType("application/json");
                CorrelationData correlationData = new CorrelationData(payload.toString());
                correlationData.setReturnedMessage(message);
    
                rabbitTemplate.convertAndSend(routingKeyName,message,correlationData);
            } catch (Exception ex) {
                logger.error("convertAndSend failed:{}", ex);
            }
        }

我这里直接将ID设置为传输内容，实际应用过程中应该设置为消息的唯一标识符。  
![在这里插入图片描述][20190827110202679.png]  
那么接下来模拟消息丢失过程。在消息传输过程中，ubbind Binding  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rhc2h1YWlnZWdlNjQy_size_16_color_FFFFFF_t_70 3]  
会发现控制台输出如下：

com.wongws.routrisk.RabbitmqConfig       : 消息丢失:exchange(routrisk_log_user_exchange),route(routrisk_log_user_routing_key),replyCode(312),replyText(NO_ROUTE),message:(Body:'"这是第 12259 个信息"' MessageProperties [headers={__ContentTypeId__=application/json, spring_returned_message_correlation=这是第 12259 个信息}, contentType=application/json, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, deliveryTag=0])

在消息传输过程中，delete exchange：  
![在这里插入图片描述][20190827111324879.png]  
以上两个操作也验证了setConfirmCallback 和 setReturnCallback的触发机制。  
当消息发送失败或者未到达时，则需要进行重发。  
基于上面的分析，我们使用一种新的方式来做到数据的不丢失。  
在rabbitTemplate异步确认的基础上  
1 在本地缓存已发送的message  
2 通过confirmCallback或者被确认的ack，将被确认的message从本地删除  
3 定时扫描本地的message，如果大于一定时间未被确认，则重发

当然了，这种解决方式也有一定的问题：  
想象这种场景，rabbitmq接收到了消息，在发送ack确认时，网络断了，造成客户端没有收到ack，重发消息。（相比于丢失消息，重发消息要好解决的多，我们可以在consumer端做到幂等）。  
重发代码如下：

public class CorrelationData extends 
    org.springframework.amqp.rabbit.support.CorrelationData {
        //消息体
        private volatile Object message;
        //交换机
        private String exchange;
        //路由键
        private String routingKey;
        //重试次数
        private int retryCount = 0;

@Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            if (ack == true) {
                logger.info("消息发送成功:correlationData({}),ack({}),cause({})", correlationData, ack, cause);
            } else {
                if (correlationData instanceof com.Configuration.CorrelationData) {
                    com.Configuration.CorrelationData messageCorrelationData = (com.Configuration.CorrelationData) correlationData;
                    String exchange = messageCorrelationData.getExchange();
                    Object message = messageCorrelationData.getMessage();
                    String routingKey = messageCorrelationData.getRoutingKey();
                    int retryCount = messageCorrelationData.getRetryCount();
                    //重试次数+1
                    ((com.Configuration.CorrelationData) correlationData).setRetryCount(retryCount + 1);
                    rabbitTemplate.convertSendAndReceive(exchange, routingKey, message, correlationData);
                }
            }
        }

此处可扩展。若是重试次数大于三次仍然发送失败，可将此消息缓存，后续人工处理或者定时处理，也可以将该消息扔入死信队列处理。

[20190826201443471.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/065360593d8644bca578102f79e75b7c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rhc2h1YWlnZWdlNjQy_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/522ba412e6a54ac2813afc1be458b02c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rhc2h1YWlnZWdlNjQy_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/057ec4239ecd4e129a48059312a74810.png
[20190827105336985.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/fd56966a52004088a73c7f50859c65cf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rhc2h1YWlnZWdlNjQy_size_16_color_FFFFFF_t_70 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/11bca37ec0ad4ad280d4a6a5c255d898.png
[20190827110202679.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/bba12f796fe845829e00a0c1bb4d2cce.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rhc2h1YWlnZWdlNjQy_size_16_color_FFFFFF_t_70 3]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/26dc1f0cd2214dc4ac0f3b5ff828f36c.png
[20190827111324879.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/e4bd0b6c067348a796eaab88d7b94682.png