RabbitMQ之延时队列

延时队列：一般来说，发布消息之后，会被交换机接收并转发给对应的队列，队列分配给消费者处理，这个过程很快秒级处理；但有时候我们希望发布完消息后，在指定的时间之后再去处理消息，这个时候就需要使用到延时队列；
虽说是延时队列，但其实也只是对死信队列的一种扩展应用罢了。

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

首先还是得创建普通队列，添加参数绑定死信队列同时设置消息过期时间，生产者发布消息到普通队列，而普通队列没有任何消费者来消费，那么消息在普通队列中存活到设定过期时间就被转发到死信队列，由死信队列的消费者消费消息，以此实现延时功能。

package com.yzm.rabbitmq_06.config;

import org.springframework.amqp.core.*;
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 DL_QUEUE = "dl_queue2";
    public static final String DL_EXCHANGE = "dl.exchange";
    public static final String DL_KEY = "dl.key2";

    @Bean
    public Queue dlQueue() {
        return QueueBuilder.durable(DL_QUEUE).build();
    }

    @Bean
    public DirectExchange dlExchange() {
        return ExchangeBuilder.directExchange(DL_EXCHANGE).build();
    }

    @Bean
    public Binding dlBinding() {
        return BindingBuilder.bind(dlQueue()).to(dlExchange()).with(DL_KEY);
    }

    // ----------------------------------------------------------------------------------------------------------

    public static final String NORMAL_EXCHANGE = "normal.exchange";
    public static final String NORMAL_QUEUE2 = "normal_queue2";
    public static final String NORMAL_KEY2 = "normal.key2";

    @Bean
    public DirectExchange exchange() {
        return ExchangeBuilder.directExchange(NORMAL_EXCHANGE).build();
    }

    // 正常队列2，添加配置
    @Bean
    public Queue queue2() {
        Map params = new HashMap<>();
        params.put("x-dead-letter-exchange", DL_EXCHANGE);//声明当前队列绑定的死信交换机
        params.put("x-dead-letter-routing-key", DL_KEY);//声明当前队列的死信路由键
        // 添加消息过期时间，10秒内还没处理完，就转发给死信队列
        params.put("x-message-ttl", 10000);
        return QueueBuilder.durable(NORMAL_QUEUE2).withArguments(params).build();
    }

    @Bean
    public Binding binding2() {
        return BindingBuilder.bind(queue2()).to(exchange()).with(NORMAL_KEY2);
    }
}

package com.yzm.rabbitmq_06.sender;

import com.yzm.rabbitmq_06.config.RabbitConfig;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

@Slf4j
@RestController
public class Sender {

    private final RabbitTemplate rabbitTemplate;

    public Sender(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    @GetMapping("/dl")
    public void dl() {
        String message = "Hello World!";
        log.info(" [ 生产者 ] Sent ==> '" + message + "'");
        rabbitTemplate.convertAndSend(RabbitConfig.NORMAL_EXCHANGE, RabbitConfig.NORMAL_KEY2, message);
    }

}

消费者，不需要对普通队列进行监听

package com.yzm.rabbitmq_06.receiver;

import com.yzm.rabbitmq_06.config.RabbitConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class Receiver {

    // 监听死信队列
    @RabbitListener(queues = RabbitConfig.DL_QUEUE)
    public void dl(Message message) {
        log.info(" [ 消费者@死信号 ] 接收到消息 ==> '" + new String(message.getBody()));
    }

}

启动，运行结果：

设置过期时间10秒，没有任何消费者对normal-queue2队列处理，10秒过后，消息转发到死信队列处理

现在这种过期时间设定就针对整个normal_queue2队列的所有消息的，还可以更细粒度的针对每一条消息设置过期时间。

package com.yzm.rabbitmq_06.config;

import org.springframework.amqp.core.*;
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 DL_QUEUE = "dl_queue2";
    public static final String DL_EXCHANGE = "dl.exchange";
    public static final String DL_KEY = "dl.key2";

    @Bean
    public Queue dlQueue() {
        return QueueBuilder.durable(DL_QUEUE).build();
    }

    @Bean
    public DirectExchange dlExchange() {
        return ExchangeBuilder.directExchange(DL_EXCHANGE).build();
    }

    @Bean
    public Binding dlBinding() {
        return BindingBuilder.bind(dlQueue()).to(dlExchange()).with(DL_KEY);
    }

    // ----------------------------------------------------------------------------------------------------------

    public static final String NORMAL_EXCHANGE = "normal.exchange";
    public static final String NORMAL_QUEUE2 = "normal_queue2";
    public static final String NORMAL_KEY2 = "normal.key2";
    public static final String NORMAL_QUEUE3 = "normal_queue3";
    public static final String NORMAL_KEY3 = "normal.key3";

    @Bean
    public DirectExchange exchange() {
        return ExchangeBuilder.directExchange(NORMAL_EXCHANGE).build();
    }

    // 正常队列2，添加配置
    @Bean
    public Queue queue2() {
        Map params = new HashMap<>();
        params.put("x-dead-letter-exchange", DL_EXCHANGE);//声明当前队列绑定的死信交换机
        params.put("x-dead-letter-routing-key", DL_KEY);//声明当前队列的死信路由键
        // 添加消息过期时间，10秒内还没处理完，就转发给死信队列
        params.put("x-message-ttl", 10000);
        return QueueBuilder.durable(NORMAL_QUEUE2).withArguments(params).build();
    }

    @Bean
    public Binding binding2() {
        return BindingBuilder.bind(queue2()).to(exchange()).with(NORMAL_KEY2);
    }
    
    // 正常队列3，添加配置
    @Bean
    public Queue queue3() {
        Map params = new HashMap<>();
        params.put("x-dead-letter-exchange", DL_EXCHANGE);//声明当前队列绑定的死信交换机
        params.put("x-dead-letter-routing-key", DL_KEY);//声明当前队列的死信路由键
        // 添加消息过期时间，10秒内还没处理完，就转发给死信队列
        // params.put("x-message-ttl", 10000);
        return QueueBuilder.durable(NORMAL_QUEUE3).withArguments(params).build();
    }
    
    @Bean
    public Binding binding3() {
        return BindingBuilder.bind(queue3()).to(exchange()).with(NORMAL_KEY3);
    }
}

package com.yzm.rabbitmq_06.sender;

import com.yzm.rabbitmq_06.config.RabbitConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.nio.charset.StandardCharsets;

@Slf4j
@RestController
public class Sender {

    private final RabbitTemplate rabbitTemplate;

    public Sender(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    @GetMapping("/dl")
    public void dl() {
        String message = "Hello World!";
        log.info(" [ 生产者 ] Sent ==> '" + message + "'");
        rabbitTemplate.convertAndSend(RabbitConfig.NORMAL_EXCHANGE, RabbitConfig.NORMAL_KEY2, message);
    }

    @GetMapping("/dl3")
    public void dl3() {
        String s = "Hello World!";
        log.info(" [ 生产者 ] Sent ==> '" + s + "'");
        //设置过期时间
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setExpiration("12000");
        Message message = new Message(s.getBytes(StandardCharsets.UTF_8), messageProperties);
        rabbitTemplate.convertAndSend(RabbitConfig.NORMAL_EXCHANGE, RabbitConfig.NORMAL_KEY3, message);
    }
}

package com.yzm.rabbitmq_06.receiver;

import com.yzm.rabbitmq_06.config.RabbitConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class Receiver {

    // 监听死信队列
    @RabbitListener(queues = RabbitConfig.DL_QUEUE)
    public void dl(Message message) {
        log.info(" [ 消费者@死信号 ] 接收到消息 ==> '" + new String(message.getBody()));
    }
}

重启。运行结果：

首页
上一篇：死信队列
下一篇：备用交换机