当前位置：首页 > news >正文

使用RabbitMQ死信交换机实现延迟消息

news 来源：原创 2024/9/20 14:34:20

文章目录

- 什么是死信交换机？
- 死信交换机实现延迟消息的思路
- 实现过程
- - 配置类
  - 消费者监听死信队列
  - 发送延迟消息
- 注意事项
- 总结

在开发过程中，我们常常会遇到需要延迟处理某些消息的场景，例如订单的支付超时处理、短信的定时发送等。本文将介绍如何使用RabbitMQ的死信交换机（Dead Letter Exchange，DLX）来实现延迟消息的处理。

什么是死信交换机？

死信交换机是一种特殊的交换机，用于处理不能被正常消费的消息。当消息在队列中出现以下几种情况时，会被转发到死信交换机：

消息被拒绝（Basic.Reject或Basic.Nack）并且requeue参数设置为false。
消息在队列中的存活时间超过了TTL（Time To Live）。
队列的最大长度已满，导致消息被丢弃。

通过配置死信交换机，我们可以将这些“死信”转发到一个特殊的队列，从而进行后续处理。

死信交换机实现延迟消息的思路

利用消息的TTL属性和死信交换机，我们可以实现延迟消息的处理。具体步骤如下：

创建一个普通交换机和队列，队列绑定一个死信交换机。
发送消息到普通队列，并设置消息的TTL。
消息过期后，会转发到死信交换机，死信交换机再将消息路由到实际处理的队列中。

实现过程

配置类

首先，我们需要配置普通交换机、队列和绑定关系：

package com.itheima.consumer.config;import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class NormalConfiguration {@Beanpublic DirectExchange normalExchange() {return ExchangeBuilder.directExchange("normal.direct").build();}@Beanpublic Queue normalQueue() {return QueueBuilder.durable("normal.queue").deadLetterExchange("dlx.direct").build();}@Beanpublic Binding normalExchangeBinding(Queue normalQueue, DirectExchange normalExchange) {return BindingBuilder.bind(normalQueue).to(normalExchange).with("hi");}
}

在上述配置中，我们创建了一个普通交换机normal.direct，以及一个普通队列normal.queue，并将队列绑定到了死信交换机dlx.direct。

消费者监听死信队列

接下来，我们需要定义一个消费者来监听死信队列，并处理延迟后的消息：

import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.core.ExchangeTypes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;@Component
public class DlxMessageListener {private static final Logger log = LoggerFactory.getLogger(DlxMessageListener.class);@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "dlx.queue", durable = "true"),exchange = @Exchange(name = "dlx.direct", type = ExchangeTypes.DIRECT),key = {"hi"}))public void listenDlxQueue1(String message) {log.info("消费者监听到dlx.queue消息：{}", message);}
}

发送延迟消息

最后，我们编写一个测试方法来发送延迟消息：

import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;@SpringBootTest
public class DelayMessageTest {@Autowiredprivate RabbitTemplate rabbitTemplate;@Testvoid testSendDelayMessage() {rabbitTemplate.convertAndSend("normal.direct", "hi", "hello", message -> {message.getMessageProperties().setExpiration("10000");return message;});}
}