rabbitMq延迟队列实现

比眉伴天荒 2022-10-20 13:41 255阅读 0赞

## 前言 ##

我们要实现延迟消息队列效果，在rabbtimq中可以通过TTL+死信的方式，把过期消息转移到死信exchange中，然后再死信exchange绑定的队列中去消费完成后期的业务逻辑。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pqY2phdmE_size_16_color_FFFFFF_t_70]  
但是这里有一个前提就是，我们TTL队列中的过期时间都是一样的，如果不一样就会如下图  
![在这里插入图片描述][20210518105959666.png]  
因为消息是排队出队的，如果前面的消息TTL时长大于后面的就会一直阻塞出队口，造成队列阻塞，而后面的消息即便是过期了也依然无法出队释放空间。  
所以私信只适合统一过期时长的消息依次入队，如果是不同过期时长的消息并且不是过期时间短的排前面就会造成阻塞或者队列爆满，从而影响实际业务和队列性能。  
因此根据不同的过期时长需要创建不同的私信队列，从而实现单一死信只处理单一时长过期的消息。但是这样也会造成死信队列创建冗余，业务累赘的问题。

参考：死信队列  
https://blog.csdn.net/zjcjava/article/details/79410137

## rabbitMq延迟队列 ##

目前Kakfa,RockMq因为mq协议也没有定义这种队列标准，因此都不支持延迟队列。RabbitMq为了解决这个情况，增加了一个延迟队列插件abbitmq\_delayed\_message\_exchange  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pqY2phdmE_size_16_color_FFFFFF_t_70 1]  
它的工作模式是消息发给延迟消息交换机，交换机Exchange根据消息头中的TTL判断如果消息过期了，它就把消息转发给绑定相应的KEY队列，然后该队列的消费者去做业务逻辑处理。

这里，延迟交换机是一个非阻塞结构的计数器，它会判断每个消息的过期时长，达到过期时长的消除才会触发转发。没有达到的就一直待在交换机中，这样不用关注你的消息过期时长如何多变和前后不一致，都可以通过它来完成。

默认情况下，它是没有安装这个插件的，需要手工安装插件。

### 安装插件abbitmq\_delayed\_message\_exchange ###

官网下载  
https://www.rabbitmq.com/community-plugins.html

下载相应版本的文件解压后它是以ez格式结尾的文件  
rabbitmq\_delayed\_message\_exchange-3.8.0.ez

在安装的环境执行命令

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

完成后，在管理界面可以看到新建exchange时多了一个Type是x-delayed-message类型  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pqY2phdmE_size_16_color_FFFFFF_t_70 2]  
如果在面板中创建延迟消息交换机，如上图，需要加上Arguments参数，而且是必须加，这样，当延迟交换机才知道消息到期后是以哪种类型发给绑定的队列。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pqY2phdmE_size_16_color_FFFFFF_t_70 3]

而，队列则普通队列绑定即可，这里就不做演示了。创建完成我们在Exchange选项卡下面向该延迟消息交换机发送一条延迟消息,上图我创建了一个5000毫秒的消息，路由参数Lazy.bill  
点击发送，客户端监听该绑定的消息队列，会发现 它不会立刻接受到消息，等一下才会接受到

## 代码实现 ##

/** * rabbitmq配置 * 实现RabbitListenerConfigurer 主要是为了消息类型转换配置实现MappingJackson2MessageConverter * */
    
    @Configuration
    public class RabbitConfig implements RabbitListenerConfigurer { 
    
    
        /** * 基础配置 * @param connectionFactory * @return */
    
        @Bean
        @Primary
        public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) { 
            RabbitTemplate template = new RabbitTemplate(connectionFactory);
    // template.setMessageConverter(new Jackson2JsonMessageConverter());
            return template;
        }
        @Bean
        public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
                ConnectionFactory connectionFactory) { 
            SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
            factory.setConnectionFactory(connectionFactory);
    // factory.setMessageConverter(new Jackson2JsonMessageConverter());
    // factory.setErrorHandler(new RabbitListenerClassCastFailErrorHandler());
            return factory;
        }
    
    
        /** * 监听器消费者消息类型转换配置,如果字符串类型这里可以不用配置 * @return */
        @Override
        public void configureRabbitListeners(RabbitListenerEndpointRegistrar registrar) { 
            registrar.setMessageHandlerMethodFactory(messageHandlerMethodFactory());
        }
    
        @Bean
        MessageHandlerMethodFactory messageHandlerMethodFactory() { 
            DefaultMessageHandlerMethodFactory messageHandlerMethodFactory = new DefaultMessageHandlerMethodFactory();
            messageHandlerMethodFactory.setMessageConverter(consumerJackson2MessageConverter());
            return messageHandlerMethodFactory;
        }
    
        @Bean
        public MappingJackson2MessageConverter consumerJackson2MessageConverter() { 
            return new MappingJackson2MessageConverter();
        }
    
    
    
        /***********延迟队列消息配置*****************/
        public static final String LAZY_EXCHANGE = "Ex.LazyExchange";
        public static final String LAZY_QUEUE = "Qu.LazyQueue";
        public static final String LAZY_KEY = "Lazy.#";
        public static final String LAZY_KEY_BILL = "Lazy.bill";
    
    
    
        @Bean
        public TopicExchange lazyExchange(){ 
            Map<String, Object> pros = new HashMap<>();
            //设置交换机支持延迟消息推送
            pros.put("x-delayed-type", "direct");
            TopicExchange exchange = new TopicExchange(LAZY_EXCHANGE, true, false,pros);
            exchange.setDelayed(true);
            return exchange;
        }
    
        @Bean
        public Queue lazyQueue(){ 
            return new Queue(LAZY_QUEUE, true);
        }
    
        @Bean
        public Binding lazyBinding(){ 
            return BindingBuilder.bind(lazyQueue()).to(lazyExchange()).with(LAZY_KEY);
        }
    
    
    
    }

消息监听消费端配置

@Slf4j
    @Component
    public class RabbitmqListener { 
     
    
        /** * 延迟队列消息处理 * @param msg * @param channel * @throws IOException */
        @RabbitListener(queues = "Qu.LazyQueue")
        @RabbitHandler
        public void onLazyMessage(Message msg, Channel channel) throws IOException { 
            long deliveryTag = msg.getMessageProperties().getDeliveryTag();
            channel.basicAck(deliveryTag, true);
            log.info("lazy receive ================================================");
            log.info("lazy receive ={}" + new String(msg.getBody()));
        }
    
    
    
    }

延迟消息发送

@Resources
    RabbitTemplate rabbitTemplate;
    
    //confirmCallback returnCallback 代码省略，请参照上一篇
        public void sendLazy(BillInfo  billInfo ){ 
            rabbitTemplate.setMandatory(true);
    // rabbitTemplate.setConfirmCallback(confirmCallback);
    // rabbitTemplate.setReturnCallback(returnCallback);
            //id + 时间戳 全局唯一
            CorrelationData correlationData = new CorrelationData("12345678909"+new Date());
    
            String  msgJson = JSON.toJSONString(billInfo);
            //发送消息时指定 header 延迟时间
            rabbitTemplate.convertAndSend(RabbitConfig.LAZY_EXCHANGE, RabbitConfig.LAZY_KEY_BILL, msgJson,
                    new MessagePostProcessor() { 
                        @Override
                        public Message postProcessMessage(Message message) throws AmqpException { 
                            //设置消息持久化,过期时长6000，这里可以根据业务类型自定义每个消息不同的过期时长,比如根据租赁合同结束时间配置不同的过期时长
    message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                            message.getMessageProperties().setDelay(6000);
                            return message;
                        }
                    }, correlationData);
        }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pqY2phdmE_size_16_color_FFFFFF_t_70]: /images/20221020/3bbdfc7597e04bd3b71b1a10c4e83704.png
[20210518105959666.png]: /images/20221020/3fa179a029ec4b8da5291457c92c9108.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pqY2phdmE_size_16_color_FFFFFF_t_70 1]: /images/20221020/311f7d3c43d944d3847ee25c496a4b69.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pqY2phdmE_size_16_color_FFFFFF_t_70 2]: /images/20221020/760874c994204505884d9bdb11e648bf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pqY2phdmE_size_16_color_FFFFFF_t_70 3]: /images/20221020/f8b527149ac14117956ee109258da4ad.png