RabbitMq延迟消费（TTL实现）

Posted 2022-11-29 ITdfq

延时消费

RabbitMQ本身并不提供延迟队列的功能，但是我们仍然可以使用RabbitMQ的 TTL（Time-To-Live） 和 DLX（Dead Letter Exchanges） 这两个扩展特性来实现延迟队列，实现消息的延迟消费和延迟重试的功能。

实现结果

• 固定时间延迟消费

  

• 指定时间消费
  

具体实现

连接配置

package com.itdfq.delay.config;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @Author: QianMo
 * @Date: 2021/10/19 18:34
 * @Description:
 */
@Configuration
public class RabbitConfig 

    @Value("$spring.rabbitmq.addresses")
    private String address;
    @Value("$spring.rabbitmq.port")
    private String port;
    @Value("$spring.rabbitmq.username")
    private String username;
    @Value("$spring.rabbitmq.password")
    private String password;
    @Value("$spring.rabbitmq.virtual-host")
    private String virtualHost;


    //连接工厂
    @Bean
    public ConnectionFactory connectionFactory() 
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setAddresses(address + ":" + port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(virtualHost);
        //TODO  消息发送确认--回调
//        connectionFactory.setPublisherConfirms(true);
        return connectionFactory;

    

    //RabbitAdmin类封装对RabbitMQ的管理操作
    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) 
        return new RabbitAdmin(connectionFactory);
    

    //使用Template
    @Bean
    public RabbitTemplate newRabbitTemplate() 
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());
        //设置监听确认mq（交换器）接受到信息
        rabbitTemplate.setConfirmCallback(confirmCallback());
        //添加监听 失败鉴定（路由没有收到）
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setReturnCallback(returnCallback());
        return rabbitTemplate;
    


    //****************生产者发送确认********************
    @Bean
    public RabbitTemplate.ConfirmCallback confirmCallback() 
        return new RabbitTemplate.ConfirmCallback() 
            @Override
            public void confirm(CorrelationData correlationData, boolean b, String s) 
                if (b) 
                    System.out.println("发送者确认发送给mq成功");
                 else 
                    System.out.println("发送者发送失败，考虑重发" + s);
                
            
        ;
    

    //****************失败通知********************
    //失败才通知，成功不通知
    @Bean
    public RabbitTemplate.ReturnCallback returnCallback() 
        return new RabbitTemplate.ReturnCallback() 
            @Override
            public void returnedMessage(Message message, int i, String replayText, String exchange, String rountingKey) 
                System.out.println("无效路由信息，需要考虑另外处理");
                System.out.println("Returned replayText:" + replayText);
                System.out.println("Returned exchange:" + exchange);
                System.out.println("Returned rountingKey:" + rountingKey);
                String s = new String(message.getBody());
                System.out.println("Returned Message:" + s);
            
        ;
    

设置延时队列

  /**
     * 延时队列
     *
     *
     * @return
     */
    @Bean
    public Queue DelayQueue() 
        Map<String, Object> params = new HashMap<>();
        // x-dead-letter-exchange 声明了队列里的死信转发到的DLX名称，
        params.put("x-dead-letter-exchange", RabbitMqConstant.IMMEDIATE_EXCHANGE);
        // x-dead-letter-routing-key 声明了这些死信在转发时携带的 routing-key 名称。
        params.put("x-dead-letter-routing-key", RabbitMqConstant.IMMEDIATE_ROUTING_KEY);
        // x-message-ttl 声明该队列死信可存活时间
        params.put("x-message-ttl", RabbitMqConstant.DELAY_TIME);
        return new Queue(RabbitMqConstant.DELAY_QUEUE, true, false, false, params);
    

设置立即消费监听

package com.itdfq.delay.message.listen;

import com.itdfq.delay.constant.RabbitMqConstant;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @Author: QianMo
 * @Date: 2021/10/21 10:29
 * @Description:
 */
@Configuration
public class DelayListenConfig 


    @Autowired
    private RabbitAdmin rabbitAdmin;

    @Autowired
    private ConnectionFactory connectionFactory;


    /********************立即消费配置**********************/

    @Bean
    public DirectExchange Exchange() 
        DirectExchange exchange = new DirectExchange(
                RabbitMqConstant.IMMEDIATE_EXCHANGE, true, false);
        exchange.setAdminsThatShouldDeclare(rabbitAdmin);
        return exchange;
    

    @Bean
    public Queue Queue() 
        Queue queue = new Queue(RabbitMqConstant.IMMEDIATE_QUEUE, true, false, false);
        queue.setAdminsThatShouldDeclare(rabbitAdmin);
        return queue;
    

    @Bean
    public Binding subscribeNotifyBinding() 
        Binding binding = BindingBuilder.bind(Queue()).to(Exchange())
                .with(RabbitMqConstant.IMMEDIATE_ROUTING_KEY);
        binding.setAdminsThatShouldDeclare(rabbitAdmin);
        return binding;
    


    @Bean
    public SimpleMessageListenerContainer container(
            @Qualifier(value = "delayRabbitmqListener") DelayRabbitmqListener delayRabbitmqListener) 
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setQueues(Queue());
        container.setMessageListener(delayRabbitmqListener);
        container.setDefaultRequeueRejected(false);
        //手动提交
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        //设置消费者ack消息的模式，默认是自动，此处设置为手动
//        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        return container;
    

---

指定时间（延时消费）

设置延时队列

   /**
     * 延时队列，可变延时
     *
     * @return
     */
    @Bean
    public Queue variableDelayQueue() 
        Map<String, Object> params = new HashMap<>();
        // x-dead-letter-exchange 声明了队列里的死信转发到的DLX名称，
        params.put("x-dead-letter-exchange", RabbitMqConstant.IMMEDIATE_EXCHANGE);
        // x-dead-letter-routing-key 声明了这些死信在转发时携带的 routing-key 名称。
        params.put("x-dead-letter-routing-key", RabbitMqConstant.IMMEDIATE_ROUTING_KEY);
        return new Queue(RabbitMqConstant.DELAY_VARIABLE_QUEUE_KEY, true, false, false, params);
    

消费者

 /**
     * 可变延时消费
     * @param msg  消息
     * @param expiration 延时时间单位 ： 毫秒
     */
    public void send(String msg,Integer expiration)
        rabbitTemplate.convertAndSend(RabbitMqConstant.DELAY_VARIABLE_EXCHANGE_KEY,
                RabbitMqConstant.DELAY_VARIABLE_ROUTING_KEY, msg,
                message -> 
                    log.info("可变延时消费发送消息: , and expiration in ms", msg, expiration);
                    message.getMessageProperties().setExpiration(expiration.toString());
                    return message;
                );