• 1. Spring Boot 集成 RabbitMQ 发送延时消息
  • 1.1. 版本说明
  • 1.2. Spring 配置
  • 1.3. 定义常量
  • 1.4. 配置交换机和队列
  • 1.5. 测试

1. Spring Boot 集成 RabbitMQ 发送延时消息

延时消息依赖于 rabbitmq-delayed-message-exchange RabbitMQ 插件实现，Github 地址，
启用该插件后可以创建 x-delayed-message 类型的交换机，该类型的交换机可以支持延时消息。
Spring 已原生支持此插件。
下载相应版本的 .ez 文件到 RabbitMQ 安装目录下的 plugins 文件夹里，执行以下命令启用插件：

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

1.1. 版本说明

1.2. Spring 配置

spring:
  application:
    name: spring-rabbit-delay-exchange-demo
  rabbitmq:
    addresses: 127.0.0.1:5672
    username: admin
    password: admin
    virtual-host: /

1.3. 定义常量

public static final String QUEUE = "spring-rabbit-delay-exchange-demo-queue";
public static final String EXCHANGE = "spring-rabbit-delay-exchange-demo-exchange";
public static final String DELAYED_ROUTING_KEY = "delayed-routing-key";

1.4. 配置交换机和队列

@Configuration
@Slf4j
public class RabbitDelayExchangeConfiguration {

    @Bean
    public Queue queue() {
        return QueueBuilder.durable(QUEUE).build();
    }

    @Bean
    public Exchange exchange() {
        return ExchangeBuilder.directExchange(EXCHANGE)
                .delayed() // 设置为延时交换机
                .durable(true)
                .build();
    }

    @Bean
    public Binding binding() {
        return BindingBuilder.bind(queue())
                .to(exchange())
                .with(DELAYED_ROUTING_KEY)
                .noargs();
    }
}

1.5. 测试

@Component
@Slf4j
public class SpringRabbitDelayExchangeDemo implements ApplicationRunner {

    @Resource
    private RabbitTemplate rabbitTemplate;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        String payload = "this is a delayed message";
        MessageProperties messageProperties = new MessageProperties();
        //延时 3 秒
        messageProperties.setDelay(3000);
        rabbitTemplate.convertAndSend(
                EXCHANGE,
                DELAYED_ROUTING_KEY,
                MessageBuilder.withBody(payload.getBytes(UTF_8))
                        .andProperties(messageProperties)
                        .build()
        );
        log.info("at {} sent a delayed message, exchange: {}, payload: {}", LocalDateTime.now(), EXCHANGE, payload);
    }

    @RabbitListener(queues = {QUEUE})
    public void listen(Message<String> message) {
        log.info(
                "at {} received a message, queue: {}, exchange: {}, payload: {}",
                LocalDateTime.now(),
                message.getHeaders().get(CONSUMER_QUEUE),
                message.getHeaders().get(RECEIVED_EXCHANGE),
                message.getPayload()
        );
    }
}

启动程序，控制台将输出：

at 2024-09-27T14:28:07.183930 sent a delayed message, exchange: spring-rabbit-delay-exchange-demo-exchange, payload: this is a delayed message
at 2024-09-27T14:28:10.261180 received a message, queue: spring-rabbit-delay-exchange-demo-queue, exchange: spring-rabbit-delay-exchange-demo-exchange, payload: this is a delayed message

可以看出在发出消息 3 秒后接收到了延时消息。