rabbitmq延迟队列的实现

第一步：下载延迟队列插件

去https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/tree/v3.9.x中下载对应版本的插件

执行 rabbitmq-plugins enable rabbitmq_delayed_message_exchange 命令，rabbitmq需重启

    com.rabbitmq
    amqp-client


    org.springframework.amqp
    spring-rabbit

package com.flexible.yiren.user.rabbitmq;


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 RabbitConfig {
    public static final String DELAY_EXCHANGE = "delay.exchange";

    public static final String DELAY_ROUTE_KEY = "delay.routeKey";

    public static final String DELAY_QUEUE = "delay.queue";

    
    @Bean
    public CustomExchange delayExchange() {
        Map arguments = new HashMap<>(1);
        arguments.put("x-delayed-type", "direct");
        return new CustomExchange(DELAY_EXCHANGE,"x-delayed-message",true,false,arguments);
    }

    
    @Bean
    public Queue delayQueue() {
        return new Queue(DELAY_QUEUE,true,false,false);
    }

    
    @Bean
    public Binding delayBind() {
        return BindingBuilder.bind(this.delayQueue()).to(this.delayExchange()).with(DELAY_ROUTE_KEY).noargs();
    }

}

package com.flexible.yiren.user.rabbitmq;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;

@Component
public class CouponHandler {


    @RabbitHandler
    @RabbitListener(queues = RabbitConfig.DELAY_QUEUE)
    public void delayConsumer(Message message) {
        System.out.println("-------"+new String(message.getBody()));
    }


}

package com.flexible.yiren.user.contorller;


import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;


@RestController
public class CouponsController {


    private final RabbitTemplate rabbitTemplate;
    
    @GetMapping("test")
    @ApiOperation("测试")
    public void test() {
        rabbitTemplate.convertAndSend(RabbitConfig.DELAY_EXCHANGE, RabbitConfig.DELAY_ROUTE_KEY, "需要发送的消息内容", msg -> {
            msg.getMessageProperties().setHeader("x-delay", 2 * 1000);//延迟两秒接收
            return msg;
        });

    }
}