Rabbitmq 定时任务 （代码实现）

Posted 2021-10-02 正怒月神

exchange分别和 queue1，queue2绑定

然后设置queue1的过期时间，以及过期后的routing_key,

而queue2 则根据过期后的routing_key拿到消息。

最后单独有一个消费者，消费queue2就可以了。

一，配置文件

import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;


/**
 * @description :
 * @auther Tyler
 * @date 2021/8/25
 */

@Configuration
public class RabbitMQConfig extends MqBaseConfig {

    //设置后，才能收到消息
    @Bean
    Jackson2JsonMessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }
    @Bean(name = "adTemplate")
    public RabbitTemplate monsterAmqpTemplate(@Qualifier("adFactory") ConnectionFactory connectionFactory) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        template.setMessageConverter(jsonMessageConverter());
        template.setUsePublisherConnection(true);
        return template;
    }


    /** 死信交换机*/
    public static final String EXCHANGE = "delay";

    /** 队列1,routing-key1 */
    public static final String QUEUE1="delay_queue1";
    public static final String ROUTINGKEY1 = "delay";

    /** 队列2,routing-key2 */
    public static final String QUEUE2="delay_queue2";
    public static final String ROUTINGKEY2 = "delay_key";


    //直连交换机
    @Bean
    public DirectExchange defaultExchange() {
        return new DirectExchange(EXCHANGE, true, false);
    }


    //队列1，过期队列
    @Bean
    public Queue queue1() {
        Map<String, Object> map = new HashMap<>();
        //绑定死信交换机
        map.put("x-dead-letter-exchange", EXCHANGE);
        //消息过期后的routing-key
        map.put("x-dead-letter-routing-key",ROUTINGKEY2);
        //过期时间
        map.put("x-message-ttl", 6000);
        return new Queue(QUEUE1, true, false, false, map); //队列持久

    }
    
    //队列2，消费队列
    @Bean
    public Queue queue2() {
        return new Queue(QUEUE2, true); //队列持久

    }
    
    //绑定exchange和queue1
    @Bean
    public Binding bindingExchangeMessage1() {
        return BindingBuilder.bind(queue1()).to(defaultExchange()).with(ROUTINGKEY1);
    }

    //绑定exchange和queue2
    @Bean
    public Binding bindingExchangeMessage2() {
        return BindingBuilder.bind(queue2()).to(defaultExchange()).with(ROUTINGKEY2);
    }

}

二，消费者

package com.tenyears.webAD.mqHandler;


import java.util.Date;
import java.util.List;
import java.util.Map;

/**
 * @description :
 * @auther Tyler
 * @date 2021/8/25
 */
@Component
@Lazy(false)
@RabbitListener(queues = RabbitMQConfig.QUEUE2)
public class AdScheduleExtendHandler {
        @RabbitHandler
    public void test1(Integer t)
    {
        System.out.println("rabbit test1 begin : "+t);
       //定时任务 这样queue1又收到了消息，然后等待队列过期，接着queue2又消费了。
    //rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE,RabbitMQConfig.ROUTINGKEY1,33);
    }
}

三，发送消息

rabbitmq页面上，直接通过exchange发送消息就可以了。