rabbitmq实现简单延时队列

下载插件地址（要注意和自己的rabbitmq的版本对应起来）： https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases

下载完之后把插件copy到mq的plugin目录下。

执行如下命令：

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

检查是否安装成功（出现如下就表示成功了）：

• 引入jar包

	
		
			org.springframework.cloud
			spring-cloud-starter-stream-rabbit
			3.0.13.RELEASE
		

• 配置yml

  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest

• 生产者

package com.itllc.mq;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;


@Component
public class MyRabbitSender {

    Logger log = LoggerFactory.getLogger(MyRabbitSender.class);

    private static final String ROUTE_KEY = "delayed_key";
    private static final String EXCHANGE_NAME = "delayed_exchange";

    @Autowired
    private RabbitTemplate rabbitTemplate;

    
    public void send2(String msg, int delay) {
        log.info("RabbitSender.send() msg = {}", msg);
        rabbitTemplate.convertAndSend(EXCHANGE_NAME, ROUTE_KEY, msg, message -> {
            message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);  //消息持久化
            message.getMessageProperties().setDelay(delay * 1000);   // 单位为毫秒
            return message;
        });
    }
}

• 消费者

package com.itllc.mq;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class MyRabbitConsume {
    Logger log = LoggerFactory.getLogger(MyRabbitConsume.class);

    @RabbitListener(queues = "delayed_queue")
    public void infoConsumption(String data) throws Exception {
        log.info("收到信息:{}",data);
        log.info("然后进行一系列逻辑处理 Thanks♪(･ω･)ﾉ");
    }
}

• 绑定交换机和路由键

package com.itllc.mq;


import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.CustomExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

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

@Configuration
public class MqDelayConfig {

    //最后经过死信队列转发后实际消费的交换机
    private static final String EXCHANGE_NAME = "delayed_exchange";
    //最后经过死信队列转发后实际消费的队列
    private static final String QUEUE_NAME = "delayed_queue";
    //最后经过死信队列转发后实际消费的路由key
    private static final String ROUTE_KEY = "delayed_key";

    
    @Bean
    CustomExchange exchange() {
        //通过x-delayed-type参数设置fanout /direct / topic / header 类型
        Map args = new HashMap<>();
        args.put("x-delayed-type", "direct");
        return new CustomExchange(EXCHANGE_NAME, "x-delayed-message",true, false,args);
    }

    
    @Bean
    public Queue queue() {
        return new Queue(QUEUE_NAME,true,false,false);
    }

    
    @Bean
    public Binding binding(CustomExchange exchange, Queue queue) {
        return BindingBuilder
                .bind(queue)
                .to(exchange)
                .with(ROUTE_KEY)
                .noargs();
    }
}

• 测试

    @Resource
    private MyRabbitSender myRabbitSender;
    // 发送延迟消息
    @PostMapping("/msg")
    @ApiOperation("发送延迟消息")
    public String sendMessage(@RequestParam("body") String body) {

        myRabbitSender.send2(body,10);
        log.info("发送延迟消息成功");
        return "SUCCESS";
    }

测试结果：