springBoot+RabbitMq 实现延时发送消息-死信方式

柔情只为你懂 2022-11-01 01:35 302阅读 0赞

*由于考虑到RabbitMq支持高并发、持久化、高可用等优点，决定从activeMq转战RabbitMq，因为前期调研不充分，结果发现RabbitMq自身竟然不支持延时消息，只能通过死信或者集成插件来实现，这里先介绍死信的方式。*

这里就不赘述Rabbit的详细用法了，就直接上代码介绍实现步骤啦。

**一.定义消息队列的常量**

public static final String ACT_QUEUE= "act_queue";//活动立即消费队列
    public static final String ACT_DELAY_QUEUE= "act_delay_queue";//活动死信消费队列
    public static final String ACT_EXCHANGE = "act_exchage";         //交换器
    public static final String ACT_DEAD_LETTER_EXCHANGE = "act_dead_letter_exchange"; //死信交换器
    public static final String ACT_ROUTING_KEY = "act_routing_key";//立即消费路由键
    public static final String ACT_DELAY_ROUTING_KEY = "act_delay_routing_key"; //延迟路由键

**二.配置队列**

*-------------------认真看，这可是完整代码*

//立即消费队列
    @Bean
    public Queue actQueue() {
        return new Queue(ACT_QUEUE,true);
    }
    
    /**
     * 延迟队列: 创建一个延迟队列, 此队列中的消息没有消费者去消费, 到了过期时间之后变成死信, 变死信之后会根据
     *           绑定的DLX和routingKey重新发送到指定交换机再到指定队列。
    */
    @Bean
    public Queue delayQueue() {
        Map<String, Object> map = new HashMap<>();
        // x-dead-letter-exchange 声明了队列里的死信转发到的DLX名称，
        map.put("x-dead-letter-exchange", ACT_EXCHANGE);
        // x-dead-letter-routing-key 声明了这些死信在转发时携带的 routing-key 名称。
        map.put("x-dead-letter-routing-key", ACT_ROUTING_KEY);
        Queue queue = new Queue(ACT_DELAY_QUEUE, true, false, false, map);
        System.out.println("----------------------------deadLetterQueue :" + queue.getArguments());
        return queue;
    }
    
    /**
     *  声明立即消费队列交换机-direct类型
     * 注：把消息投递到那些binding key与routing key完全匹配的队列中。
     * */
    @Bean
    public DirectExchange actExchage(){
        // 一共有三种构造方法，可以只传exchange的名字， 第二种，可以传exchange名字，是否支持持久化，是否可以自动删除，
        //第三种在第二种参数上可以增加Map，Map中可以存放自定义exchange中的参数
        return new DirectExchange (ACT_EXCHANGE,true, false);
    }
    
    //声明死信队列交换机-direct类型
    @Bean
    public DirectExchange deadLetterExchange() {
        return new DirectExchange(ACT_DEAD_LETTER_EXCHANGE, true, false);
    }
    
    /**
     * @description   绑定队列与交换机
     *                  把立即消费的队列和立即消费交换机绑定, immediate_exchange, 路由键：immediate_routing_key
     * @author        songchengye
     * @date          2020/10/24 13:24
     */
    @Bean
    public Binding actBinding() {
        return BindingBuilder.bind(actQueue()).to(actExchage()).with(ACT_ROUTING_KEY);
    }
    
    /**
     * @description   死信队列的交换机绑定:把延迟消费的队列和死信交换机绑定, immediate_dead_exchange, 路由键：delay_routing_key
     * @author        songchengye
     * @date          2020/10/24 13:24
    */
    @Bean
    public Binding queueDeadBinding() {
        return BindingBuilder.bind(delayQueue()).to(deadLetterExchange()).with(ACT_DELAY_ROUTING_KEY);
    }

三.编写消息发送者

import com.alibaba.fastjson.JSONObject;
    import com.jeesite.modules.sp.entity.SpAct;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.AmqpException;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.core.MessagePostProcessor;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    import java.util.Date;
    
    
    @Service
    public class MQActSender {
    
    private static Logger log = LoggerFactory.getLogger(MQActSender.class);
    
    @Autowired
    RabbitTemplate rabbitTemplate;
    
    //发送延时消息
    public void sendDelayCancelAct(Object message){
        log.info("进入活动消息生产者-----------sendDelayCancelAct---------------发送活动延时消息:"+message.toString());
        SpAct spAct = (SpAct) message;
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("test","测试发送消息到死信队列");
        String time = "60000";//延时时间，这样得到的值是毫秒级,rabbitMQ默认是毫秒
        log.info("----sendDelayCancelAct---------------------------活动消息延迟时间:"+time);
        rabbitTemplate.convertAndSend(MQConfig.ACT_DEAD_LETTER_EXCHANGE, MQConfig.ACT_DELAY_ROUTING_KEY, jsonObject,
                new MessagePostProcessor() {
                    @Override
                    public Message postProcessMessage(Message message) throws AmqpException {
                        //设置消息持久化
                        //message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                        message.getMessageProperties().setExpiration(time);
                        return message;
                    }
                });
    }

四.编写消息消费者

import com.alibaba.fastjson.JSONObject;
    import com.jeesite.modules.sp.api.constant.Dict;
    import com.jeesite.modules.sp.entity.SpAct;
    import com.jeesite.modules.sp.entity.SpActEnter;
    import com.jeesite.modules.sp.service.SpActEnterService;
    import com.jeesite.modules.sp.service.SpActService;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    import org.springframework.stereotype.Service;
    import java.util.Date;
    
    
    @Service
    @Component
    public class MQActReceiver {
    
    @Autowired
    private SpActService spActService;
    
    @Autowired
    private SpActEnterService spActEnterService;
    
    
    private static Logger log = LoggerFactory.getLogger(MQActReceiver.class);
    
    @RabbitListener(queues  = MQConfig.ACT_QUEUE)
    public void receive(JSONObject message){
        log.info("进入消息消费者----------------------------监听活动消息 = " + message.toJSONString());
        try {
            String test= message.getString("test").toString();
            log.info("---------------------------MQActReceiver--------监听到的消息："+test
        } catch (Exception e) {
            e.printStackTrace();
        }
    
    }

到此，基本就实现了延时发送，但由于队列先进先出的问题，会出现消息阻塞的问题，例如A消息是10分钟，即使B消息是5分钟，那么B消息也只能等A消息消费以后才能被消费。

大致流程如下图：  
*死信队列是没有自己的消费者的。*  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mzk0NTk4Mw_size_16_color_FFFFFF_t_70_pic_center]

五、后记（该方法的缺陷）

**RabbitMq TTL 队列过期时间长的消息阻塞后面过期时间短的消息**

利用RabbitMq的TTL time-to-live 来实现延时消费。

延时时间设置到每条消息上的。而不是给队列的。

实现方式为消息存活时间为动态用户页面可配置的。

这就导致了一个问题：

先用一条消息的存活时间是1天。后面又进了一条消息存活时间是1小时。

结果一小时到了，发现这条消息并没有被转发到消费延时过期消息的队列。

原因是尽管ttl是设给每条消息的。但是本质上，所有延时消息都还在一个队列里，对它过期时间的检测也是从头部开始的。

它不会检测每一条消息是否过期。而是顺序检测。

如果first in的消息过期时间很长，会导致它阻塞后进的消息。

不仅无法实现真正的过期时间。还会导致，一个大的过期时间的先进的消息，会堆积一堆后进的过期时间短的消息。

当先进的这条消息过期转发时，后面的消息会瞬间都被转发。造成消费延时过期消息的队列瞬间被转发进来大量消息。

造成消费压力。

**所以rabbitmq ttl适用于过期时间一致的延时消费。**

**对于动态的延时可变的情况，还是考虑定时任务实现。**

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mzk0NTk4Mw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221024/1c19a7f4ea1a44eb92ec6403355f1610.png