RabbitMq进阶——延迟队列使用

使用场景

1、定时任务，比如订单超过30分钟自动取消

2、监听任务，比如每隔10秒，监听多个子任务是否已经完成

实现方案

本人讲述spring boot框架下的实现方案

延迟队列绑定

import org.springframework.amqp.core.*;

    /**
     * 延时队列设置
     * 设置延时队列的过期时间为2秒钟
     * 2秒之后，延时队列将消息发送给消费队列
     */
    @Bean
    public Queue testDelayQueue(){
        Map args = new HashMap();
        args.put("x-dead-letter-exchange", AMQPConstants.TEST_EXCHANGE);
        args.put("x-dead-letter-routing-key",AMQPConstants.TEST);
        args.put("x-message-ttl",2000);
        return QueueBuilder.durable(AMQPConstants.TEST_DELAY).withArguments(args).autoDelete().build();
    }

    /**
     * 延时交换机
     */
    @Bean
    public Exchange testDelayExchange(){
        return ExchangeBuilder.directExchange(AMQPConstants.TEST_DELAY_EXCHANGE).autoDelete().build();
    }

    /**
     * 延时队列与延时交换机进行绑定
     */
    @Bean
    public Binding testDelayBindDelay(Queue testDelayQueue, Exchange testDelayExchange){
        return BindingBuilder.bind(testDelayQueue).to(testDelayExchange).with(AMQPConstants.TEST_DELAY).noargs();
    }

发送端发送TEST_DELEY队列

// 发送端发送到延迟队列
amqpTemplateHelper.send(AMQPConstants.TEST_DELAY, data);

消费端为TEST队列，接受到的消息为延迟队列时间到期的消息

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Argument;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class TestReceiver {

    @Autowired
    private TestService service;

    @RabbitListener(bindings = @QueueBinding(value = @Queue(value = AMQPConstants.TEST,  autoDelete = "true"),
                    exchange = @Exchange(value = AMQPConstants.TEST_EXCHANGE, autoDelete = "true"),
                    key = AMQPConstants.TEST))
    public void process(String data) {
        service.process(data);
    }
}