.NET 6+EasyNetQ（RabbitMQ 延迟消息指南）

Connor usdt钱包 2022-10-27 310 0

↓推荐关注↓

背景

最近遇到一个比较特殊需求期货指南，需要修改一个的RabbitMQ消费者，以实现在消费某种特定的类型消息时，延迟1小时再处理，几个需要注意的点：

延迟是以小时为单位

不是所有消息都延迟消费期货指南，只延迟特定类型的消息

只在第一次消费时延迟1小时期货指南，容错机制产生的重新消费（也即消息消费失败，多次进入延迟队列重试），则不再延迟1小时

消费者消费过程中可能会重启

延迟是以小时为单位

不是所有消息都延迟消费期货指南，只延迟特定类型的消息

只在第一次消费时延迟1小时期货指南，容错机制产生的重新消费（也即消息消费失败，多次进入延迟队列重试），则不再延迟1小时

消费者消费过程中可能会重启

考虑到这几点期货指南，我们需要一个标识以及持久化，不能简单使用Thread.Sleep或者Task.Delay；

下面开始演示在不引入其它框架资源的前提下，利用现有的RabbitMQ来实现这个需求期货指南。

准备

如果没有可用的RabbitMQ测试环境期货指南，推荐使用docker本地搭建

dockerrun-it--rm--namerabbitmq-p5672 :5672-p15672 :15672rabbitmq:3.10-management

项目搭建

创建解决方案RabbitMQDemo期货指南，并添加一个.Net6控制台程序Producer作为生产者，

mkdir RabbitMQDemo

cdRabbitMQDemo

dotnet new sln -n RabbitMQDemo

mkdir src

展开全文

cdsrc

dotnet new console -n Producer

cdProducer

dotnet add package EasyNetQ -s

dotnet add package Newtonsoft.Json -s

cd../..

dotnet sln add ./src/Producer/Producer.csproj

我们给Producer项目添加了两个包 ——EasyNetQ是用来简便RabbitMQ操作，添加Newtonsoft.Json则是因为EasyNetQ从v7版本开始移除了对前者的依赖，需要使用者自行添加期货指南。

接下来定义消息的数据结构期货指南，添加一个类库Core到解决方案，

cdsrc

dotnet new classlib --name Core

cd..

dotnet sln add ./src/Core/Core.csproj

添加如下OrderNotification类，后面我们根据消息的 Type的值来确定是正常消费还是延迟消费期货指南。

namespaceCore

publicclassOrderNotification

publicstringOrderId { get; set; }

publicintType { get; set; }

publicDateTime DateCreation { get; set; }

生产者

在Producer项目里期货指南，声明队列orders.notification，绑定到同名交换机，然后向该交换机发送OrderNotification类型的数据，

实际项目中期货指南，我们很少直接发消息到队列，都是发送到交换机，这个项目虽然只是demo，但也遵循这个原则

完整代码如下：

usingCore;

usingEasyNetQ;

usingEasyNetQ.Topology;

varbus = RabbitHutch.CreateBus( "host=localhost;port=5672;virtualHost=/;username=guest;password=guest;requestedHeartbeat=10");

//声明交换机

varsourceExchange = awaitbus.Advanced.ExchangeDeclareAsync(name: "orders.notification", ExchangeType.Direct);

//声明队列

varsourceQueue = awaitbus.Advanced.QueueDeclareAsync(name: "orders.notification");

//绑定

awaitbus.Advanced.BindAsync(sourceExchange, sourceQueue, "");

Console.WriteLine( "按Ctrl + C 暂停发送期货指南，任意键恢复发送");

Console.TreatControlCAsInput = true;

while( true)

Random random = new;

varorderId = Guid.NewGuid.ToString;

vartype = random.Next( 1, 3);

awaitbus.Advanced.PublishAsync(sourceExchange, "", true, newMessage<OrderNotification>( newOrderNotification { OrderId = orderId, Type = type, DateCreation = DateTime.Now }));

Console.WriteLine( $" {DateTime.Now}:消息（OrderId: {orderId}期货指南，Type： {type}）已发送" );

Thread.Sleep( 1000);

运行Producer项目期货指南，可以看到消息正在不停的发送

打开RabbitMQ后台期货指南，名orders.notification的队列和交换机已经创建好且相互绑定，队列里已经有我们刚刚发送的消息

下面我们要做的就是将队列orders.notification里Type为1的消息延迟消费，其它则正常消费期货指南。

延迟消费

使用死信交换机实现

原理就是在声明一个队列时，给它配置死信交换机（Dead Letter Exchanges，简称DLX）策略，对应参数为x-dead-letter-exchange，这种队列处理带设置了过期时间属性（Properties.expiration）的消息时，在消息到期时，会自动将消息投递到事先配置好的死信交换机上期货指南。

期货指南我们解决方案增加一个控制台类型的消费者项目DLXConsumer

cdsrc

dotnet new console -n DLXConsumer

cdDLXConsumer

dotnet add package EasyNetQ -s

dotnet add package Newtonsoft.Json -s

cd../..

dotnet sln add ./src/DLXConsumer/DLXConsumer.csproj

和生产者类似期货指南，实现消费者我们也创建一对同名的交换机和队列 orders.notification_dlx ，用于接收转发过来延迟消息，同时将该队列的死信交换机设置为 orders.notification ；

消费消息时，为了消息是不是已经延迟过，可以在消息头里添加一个自定义参数 biz-delayed ，在将需要延迟处理的消息发送到 orders.notification_dlx 交换机之前，除了设置过期时间，也同时将 biz-delayed 设置为1，后续再消费该消息时，读取该值，不至于陷入死循环期货指南。完整代码如下

usingCore;

usingEasyNetQ;

usingEasyNetQ.Topology;

varbus = RabbitHutch.CreateBus( "host=localhost;port=5672;virtualHost=/;username=guest;password=guest;requestedHeartbeat=10");

varsourceExchange = awaitbus.Advanced.ExchangeDeclareAsync(name: "orders.notification", ExchangeType.Direct);

varsourceQueue = awaitbus.Advanced.QueueDeclareAsync(name: "orders.notification");

vardlxExchange = awaitbus.Advanced.ExchangeDeclareAsync(name: "orders.notification_dlx", ExchangeType.Direct);

vardlxQueue = awaitbus.Advanced.QueueDeclareAsync(name: "orders.notification_dlx", configure => configure.WithDeadLetterExchange(sourceExchange));

awaitbus.Advanced.BindAsync(dlxExchange, dlxQueue, "");

bus.Advanced.Consume<OrderNotification>(sourceQueue, OrderNotificationHandler);

Console.ReadLine;

asyncTask OrderNotificationHandler( IMessage<OrderNotification> message, MessageReceivedInfo msgInfo)

Console.WriteLine( $" {DateTime.Now}: 开始消费 OrderId: {message.Body.OrderId}Type： {message.Body.Type}" );

if(message.Body.Type == 1&& !message.Properties.Headers.ContainsKey( "biz-delayed"))

message.Properties.Headers.Add( "biz-delayed", 1);

message.Properties.Expiration = TimeSpan.FromHours( 1);

awaitbus.Advanced.PublishAsync(dlxExchange, "", true, message);

Console.WriteLine( $" {DateTime.Now}:OrderId: {message.Body.OrderId}Type： {message.Body.Type}已延迟消费" );

else

//假装在消费

Thread.Sleep( 1000);

Console.WriteLine( $" {DateTime.Now}:OrderId: {message.Body.OrderId}Type： {message.Body.Type}已成功消费" );

上述代码中，EasyNetQ设置队列死信交换机的API为WithDeadLetterExchange，设置消息过期时间的API为Properties.Expiration期货指南。

运行DLXConsumer项目期货指南，可以看到Type为1的消息被延迟，其它则被正常消费

打开RabbitMQ后台确认，原本orders.notification里的消息已经被消费掉了，同时多了一个orders.notification_dlx队列，并且orders.notification_delay队列相比orders.notification多了一个DLX标签，Type为1的消息就是被转移该队列期货指南。

进入orders.notification_delay队列期货指南，交换机与队列正常绑定，x-dead-letter-exchange也已被设置

使用延迟交换机实现

使用延迟交换机，需要RabbitMQ服务器安装rabbitmq_delayed_message_exchange插件，原理是投递到延迟交换机的消息，会延迟指定时间（x-delay参数设置）后，自动投递到该交换机绑定的另一交换机上期货指南。直接看代码。

docker环境安装rabbitmq_delayed_message_exchange插件

这里介绍下docker环境如何安装rabbitmq_delayed_message_exchange插件期货指南，首先在github 目录下，命令如下

docker cp {rabbitmq_delayed_message_exchange文件路径} {rabbitmq容器id}:/plugins

docker exec-it {rabbitmq容器id} rabbitmq-plugins enablerabbitmq_delayed_message_exchange

以我本机为例，插件启用成功期货指南。

下面给解决方法添加一个DMConsumer项目期货指南。

cdsrc

dotnet new console -n DMConsumer

cdDMConsumer

dotnet add package EasyNetQ -s

dotnet add package Newtonsoft.Json -s

cd../..

dotnet sln add ./src/DMConsumer/DMConsumer.csproj

DMConsumer完整实现如下

usingCore;

usingEasyNetQ;

usingEasyNetQ.Topology;

varbus = RabbitHutch.CreateBus( "host=localhost;port=5672;virtualHost=/;username=guest;password=guest;requestedHeartbeat=10");

varsourceExchange = awaitbus.Advanced.ExchangeDeclareAsync(name: "orders.notification", ExchangeType.Direct, durable: true, autoDelete: false);

varsourceQueue = awaitbus.Advanced.QueueDeclareAsync(name: "orders.notification");

vardmExchange = awaitbus.Advanced.ExchangeDeclareAsync(name: "orders.notification_dm", configure => configure.AsDelayedExchange(ExchangeType.Direct));

//两个交换机绑定

awaitbus.Advanced.BindAsync(dmExchange, sourceExchange, "");

bus.Advanced.Consume<OrderNotification>(sourceQueue, OrderNotificationHandler);

Console.ReadLine;

asyncTask OrderNotificationHandler( IMessage<OrderNotification> message, MessageReceivedInfo msgInfo)

Console.WriteLine( $" {DateTime.Now}: 开始消费 OrderId: {message.Body.OrderId}Type： {message.Body.Type}" );

if(message.Body.Type == 1&& !message.Properties.Headers.ContainsKey( "biz-delayed"))

message.Properties.Headers[ "biz-delayed"] = 1;

message.WithDelay(TimeSpan.FromHours( 1));

awaitbus.Advanced.PublishAsync(dmExchange, "", true, message);

Console.WriteLine( $" {DateTime.Now}:OrderId: {message.Body.OrderId}Type： {message.Body.Type}已延迟消费" );

else

//假装在消费

//Thread.Sleep(1000);

Console.WriteLine( $" {DateTime.Now}:OrderId: {message.Body.OrderId}Type： {message.Body.Type}已成功消费" );

相比于 DLXConsumer 期货指南， DMConsumer 里，我们不需要额外的队列，只需要创建 orders.notification_dm 交换机即可，同时直接将交换机绑定到 orders.notification 交换机， EasyNetQ 里使用 AsDelayedExchange 指示该交换机为延迟交换机，使用 WithDelay 设置消息延迟时间通过查看 EasyNetQ 源码 , 它封装延迟交换机的设置

启动Producer再生成一些数据期货指南，然后运行DMConsumer,效果和DLXConsumer一样

打开RabbitMQ后台期货指南，可以看到多了一个类型为x-delayed-message的orders.notification_dm交换机，带有DM和Args两个标签

进入交换机，可以看到里面已经存储了13条消息期货指南。

总结

自此期货指南，利用队列的死信交换机策略和利用rabbitmq_delayed_message_exchange插件实现RabbitMQ消息延迟已经介绍完毕，下面是.NET 6 demo完整的项目结构

其实除了这两种，EasyNetQ也有一个调度器（Scheduler）可以实现延迟消息，但似乎需要依赖数据库，不是主流的做法不推荐使用期货指南。

转自：几秋

链接： cnblogs.com/netry/p/rabbitmq-delayed-message-dotnet.html

转自：几秋

链接： cnblogs.com/netry/p/rabbitmq-delayed-message-dotnet.html

- EOF -

点击标题可跳转