南京网站制作千,网站模版怎么编辑器,如何提升做网站的效率,wordpress 授权登录文章目录 什么是死信交换机#xff1f;死信交换机实现延迟消息的思路实现过程配置类消费者监听死信队列发送延迟消息 注意事项总结 在开发过程中#xff0c;我们常常会遇到需要延迟处理某些消息的场景#xff0c;例如订单的支付超时处理、短信的定时发送等。本文将介绍如何使… 文章目录 什么是死信交换机死信交换机实现延迟消息的思路实现过程配置类消费者监听死信队列发送延迟消息 注意事项总结 在开发过程中我们常常会遇到需要延迟处理某些消息的场景例如订单的支付超时处理、短信的定时发送等。本文将介绍如何使用RabbitMQ的死信交换机Dead Letter ExchangeDLX来实现延迟消息的处理。 什么是死信交换机
死信交换机是一种特殊的交换机用于处理不能被正常消费的消息。当消息在队列中出现以下几种情况时会被转发到死信交换机
消息被拒绝Basic.Reject或Basic.Nack并且requeue参数设置为false。消息在队列中的存活时间超过了TTLTime 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;});}
}在上述测试方法中我们向普通队列发送了一条消息并设置其TTL为10000毫秒即10秒。当消息在普通队列中存活超过10秒后会被转发到死信交换机然后由消费者监听并处理。
注意事项
需要注意的是RabbitMQ的消息过期是基于追溯方式来实现的也就是说当一个消息的TTL到期以后不一定会立即被移除或投递到死信交换机而是在消息恰好处于队首时才会被处理。当队列中消息堆积很多的时候过期消息可能不会被按时处理因此你设置的TTL时间不一定准确。
总结
通过以上配置和代码我们实现了使用RabbitMQ死信交换机来处理延迟消息。利用死信交换机的机制我们可以方便地实现各种复杂的消息处理场景提高系统的灵活性和可靠性。
