netcore下RabbitMQ队列、死信队列、延时队列及小应用

作者：ouyan1985_998 | 来源：互联网 | 2023-10-12 13:36

原标题：netcore下RabbitMQ队列、死信队列、延时队列及小应用关于安装rabbitmq这里一笔掠过了。下面进入正题：1.新建aspnetcor

原标题：netcore下RabbitMQ队列、死信队列、延时队列及小应用

关于安装rabbitmq这里一笔掠过了。

下面进入正题：

1.新建aspnetcorewebapi空项目，NormalQueue,删除controllers文件夹已经无关的文件，这里为了偷懒不用console控制台：

public class Program
{
public static void Main(string[] args)
{
var builder = WebApplication.CreateBuilder(args);
// Add services to the container.

builder.Services.AddControllers();
// Learn more about configuring Swagger/OpenAPI at https://aka.ms/aspnetcore/swashbuckle
builder.Services.AddEndpointsApiExplorer();
builder.Services.AddSwaggerGen();
builder.Services.AddHostedService();
builder.Services.AddHostedService();
builder.Services.AddHostedService();
var app = builder.Build();
/文章来源站点https://www.yii666.com// Configure the HTTP request pipeline.
if (app.Environment.IsDevelopment())
{
app.UseSwagger();
app.UseSwaggerUI();
}
app.MapGet("/normal/{message}", ([FromRoute] string message) =>
{
ConnectionFactory factory = new ConnectionFactory();
factory.HostName = "localhost";
factory.Port = 5672;
using (IConnection cOnnection= factory.CreateConnection())
{
using (IModel channel = connection.CreateModel())
{
var queueName = "rbTest202301";
channel.QueueDeclare(queueName, true, false, false, null);
{
string sendMessage = string.Format("Message_{0}", message);
byte[] buffer = Encoding.UTF8.GetBytes(sendMessage);
IBasicProperties basicProperties = channel.CreateBasicProperties();
basicProperties.DeliveryMode = 2; //持久化 1=非持久化
channel.BasicPublish("", queueName, basicProperties, buffer);
Console.WriteLine("消息发送成功：" + sendMessage);
}
}
}
});
app.MapGet("/deadletterexchange/{message}",([FromRoute] string message) =>{
DeadLetterExchange.Send(message);
});
app.MapGet("/delayexchange/{message}", ([FromRoute] string message) => {
DelayExchange.SendMessage(message);
});
app.UseHttpsRedirection();
app.UseAuthorization();
app.MapControllers();
app.Run();
}
}

大概的介绍一下program文件：

这里有三个mini控制器，从这里发送对应的消息到rabbitmq

"/normal/{message}" 普通队列，

"/deadletterexchange/{message}" 死信队列

"/deadletterexchange/{message}" 延时队列

　　　　　　builder.Services.AddHostedService();
builder.Services.AddHostedService();
builder.Services.AddHostedService();

这里就是消费的服务，注册成HostedService。

ConsumerService代码如下：

DeadLetterExchangeConsuerService代码如下：

public class DeadLetterExchangeConsuerService : BackgroundService
{
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
Console.WriteLine("RabbitMQ消费端死信队列开始工作");
while (!stoppingToken.IsCancellationRequested)
{
DeadLetterExchange.Consumer();
await Task.Delay(5000);
}
}
}

public class DeadLetterExchange
{
public static string dlxExchange = "dlx.exchange";
public static string dlxQueueName = "dlx.queue";
static string exchange = "direct-exchange";
static string queueName = "queue_Testdlx";
static string dlxExchangeKey = "x-dead-letter-exchange";
static string dlxQueueKey = "x-dead-letter-rounting-key";
public static void Send(string message)
{
using (var cOnnection= new ConnectionFactory() { HostName = "localhost", Port = 5672 }.CreateConnection())
{
using(var channel = connection.CreateModel())
{

channel.ExchangeDeclare(exchange, ExchangeType.Direct, true, false); //创建交换机
channel.QueueDeclare(queueName, true, false, false,new Dictionary<string, object>
{
{ dlxExchangeKey,dlxExchange },
{dlxQueueKey,dlxQueueName }
}); // 创建队列
channel.QueueBind(queueName, exchange, queueName);
var properties = channel.CreateBasicProperties();
properties.Persistent= true;//持久化
channel.BasicPublish(exchange,queueName,properties,Encoding.UTF8.GetBytes(message));
Console.WriteLine($"向队列:{queueName}发送消息:{message}");
}
}
}

public static void Consumer()
{
var cOnnection= new ConnectionFactory() { HostName = "localhost", Port = 5672 }.CreateConnection();
var channel = connection.CreateModel();
channel.ExchangeDeclare(dlxExchange, ExchangeType.Direct, true, false); //创建sixin交换机
channel.QueueDeclare(dlxQueueName, true, false, false); // 创建sixin队列
channel.QueueBind(dlxQueueName, dlxExchange, dlxQueueName); //绑定sixin队列sixin交换机

channel.ExchangeDeclare(exchange, ExchangeType.Direct, true, false); //创建交换机
channel.QueueDeclare(queueName, true, false, false, new Dictionary<string, object>
{
{ dlxExchangeKey,dlxExchange },
{dlxQueueKey,dlxQueueName }
}); // 创建队列
channel.QueueBind(queueName, exchange, queueName);
var cOnsumer= new EventingBasicConsumer(channel);
channel.BasicQos(0, 1, false);
consumer.Received += (model, ea) =>
{
var message = Encoding.UTF8.GetString(ea.Body.ToArray());
Console.WriteLine($"队列{queueName}消费消息:{message},不做ack确认");
channel.BasicNack(ea.DeliveryTag, false, requeue: false);
};
channel.BasicConsume(queueName, autoAck: false, consumer);
}
}

DelayExchangeConsumerService代码如下：

public class DelayExchangeConsumerService : BackgroundService
{
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
Console.WriteLine("RabbitMQ消费端延迟队列开始工作");
while (!stoppingToken.IsCancellationRequested)
{

DelayExchange.Consumer();
await Task.Delay(5000);
}
}
}

public class DelayExchange
{
public static void SendMessage(string message)
{
//死信交换机
string dlxexChange = "dlx.exchange";
//死信队列
string dlxQueueName = "dlx.queue";
//消息交换机
string exchange = "direct-exchange";
//消息队列
string queueName = "delay_queue";
using (var cOnnection= new ConnectionFactory() { HostName = "localhost", Port = 5672 }.CreateConnection())
{
using (var channel = connection.CreateModel())
{
////创建死信交换机
//channel.ExchangeDeclare(dlxexChange, type: ExchangeType.Direct, durable: true, autoDelete: false);
////创建死信队列
//channel.QueueDeclare(dlxQueueName, durable: true, exclusive: false, autoDelete: false);
////死信队列绑定死信交换机
//channel.QueueBind(dlxQueueName, dlxexChange, routingKey: dlxQueueName);
// 创建消息交换机
channel.ExchangeDeclare(exchange, type: ExchangeType.Direct, durable: true, autoDelete: false);
//创建消息队列,并指定死信队列，和设置这个队列的消息过期时间为10s
channel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false, arguments:
new Dictionary<string, object> {
{ "x-dead-letter-exchange",dlxexChange}, //设置当前队列的DLX(死信交换机)
{ "x-dead-letter-routing-key",dlxQueueName}, //设置DLX的路由key，DLX会根据该值去找到死信消息存放的队列
{ "x-message-ttl",10000} //设置队列的消息过期时间
});
//消息队列绑定消息交换机
channel.QueueBind(queueName, exchange, routingKey: queueName);
var properties = channel.CreateBasicProperties();
properties.Persistent = true;
//properties.Expiration = "5000";发布消息,延时5s
//发布消息
channel.BasicPublish(exchange: exchange,
routingKey: queueName,
basicProperties: properties,
body: Encoding.UTF8.GetBytes(message));
Console.WriteLine($"{DateTime.Now},向队列:{queueName}发送消息:{message}");
}
}
}
public static void Consumer()
{
//死信交换机
string dlxexChange = "dlx.exchange";
//死信队列
string dlxQueueName = "dlx.queue";
var cOnnection= new ConnectionFactory() { HostName = "localhost", Port = 5672 }.CreateConnection();
{
//创建信道
var channel = connection.CreateModel();
{
//创建死信交换机
channel.ExchangeDeclare(dlxexChange, type: ExchangeType.Direct, durable: true, autoDelete: false);
//创建死信队列
channel.QueueDeclare(dlxQueueName, durable: true, exclusive: false, autoDelete: false);
//死信队列绑定死信交换机
channel.QueueBind(dlxQueueName, dlxexChange, routingKey: dlxQueueName);
var consumwww.yii666.comer = new EventingBasicConsumer(channel);
channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: true);
consumer.Received += (model, ea) =>
{
//处理业务
var message = Encoding.UTF8.GetString(ea.Body.ToArray());
Console.WriteLine($"{DateTime.Now}，队列{dlxQueueName}消费消息:{message}");
channel.BasicAck(ea.DeliveryTag, false);
};
channel.BasicConsume(dlxQueueName, autoAck: false, consumer);
}
}
}
}

延时队列实际应用场景可能比较复杂，比如每条消息的过期时间不一样，收到的消息的顺序有可能会乱掉。这些不做深究，自行百度。

关于死信队列常见应用场景之一下单，支付，支付超时的各种场景，下面通过一个简单的例子模拟一下

同样的新建一个空的webapi项目DeadLetterQueue，

program代码如下：

public class Program
{
public static void Main(string[] args)
{
var builder = WebApplication.CreateBuilder(args);
// Add services to the container.

builder.Services.AddControllers();
// Learn more about configuring Swagger/OpenAPI at https://aka.ms/aspnetcore/swashbuckle
builder.Services.AddEndpointsApiExplorer();
builder.Services.AddSwaggerGen();
builder.Services.AddHostedService();
builder.Services.AddHostedService();
var app = builder.Build();
// Configure the HTTP request pipeline.
if (app.Environment.IsDevelopment())
{
app.UseSwagger();
app.UseSwaggerUI();
}
app.MapGet("/normal/{message}", ([FromRoute] string message) =>
{
ConnectionFactory factory = new ConnectionFactory();
factory.HostName = "localhost";
factory.Port = 5672;
using (IConnection cOnnection= factory.CreateConnection())
{
using (IModel channel = connection.CreateModel())
{
var queueName = "rbTest2023010";

//channel.ExchangeDeclare("exchange.dlx", ExchangeType.Direct, true);
//channel.QueueDeclare("queue.dlx", true, false, false, null);
channel.ExchangeDeclare("exchange.normal", ExchangeType.Fanout, true);
channel.QueueDeclare(queueName, true, false, false,
new Dictionary<string, object>
{
{ "x-message-ttl" ,10000},
{"x-dead-letter-exchange","exchange.dlx" },
{"x-dead-letter-routing-key","routingkey" }
}
);

channel.QueueBind(queueName, "exchange.normal", "");
{
string sendMessage = string.Format("Message_{0}", message);
byte[] buffer = Encoding.UTF8.GetBytes(sendMessage);
IBasicProperties basicProperties = channel.CreateBasicProperties();
basicProperties.DeliveryMode = 2; //持久化 1=非持久化
channel.BasicPublish("exchange.normal", queueName, basicProperties, buffer);
Console.WriteLine($"{DateTime.Now}消息发送成功：{sendMessage}" );
}
}
}
});
app.UseHttpsRedirection();
app.UseAuthorization();
app.MapControllers();
app.Run();
}
}

下单后消费代码ConsumerService如下

public class ConsumerService : BackgroundService
{
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
Console.WriteLine("normal Rabbitmq消费端开始工作!");
while (!stoppingToken.IsCancellationRequested)
{
ConnectionFactory factory = new ConnectionFactory();
factory.HostName = "localhost";
factory.Port = 5672;
IConnection connection = factory.CreateConnection();
{
IModel channel = connection.CreateModel();
{
var queueName = "rbTest2023010";
channel.ExchangeDeclare("exchange.normal", ExchangeType.Fanout, true);
channel.QueueDeclare(queueName, true, false, false, new Dictionary<string, object>
{
{ "x-message-ttl" ,10000},
{"x-dead-letter-exchange","exchange.dlx" },
{"x-dead-letter-routing-key","routingkey" }
});
channel.QueueBind(queueName, "exchange.normal", "");
//输入1，那如果接收一个消息，但是没有应答，则客户端不会收到下一个消息
channel.BasicQos(0, 1, false);
//在队列上定义一个消费者
var cOnsumer= new EventingBasicConsumer(channel);
channel.BasicConsume(queueName, false, consumer);
consumer.Received += (ch, ea) =>
{
byte[] bytes = ea.Body.ToArray();
string str = Encoding.UTF8.GetString(bytes);
Console.WriteLine($"{DateTime.Now}来自死信队列获取的消息: {str.ToString()}");
//回复确认
if (str.Contains("跳过")) //假设超时不处理，留给后面deadconsumerservice处理
{
Console.WriteLine($"{DateTime.Now}来自死信队列获取的消息: {str.ToString()},该消息被拒绝");
channel.BasicNack(ea.DeliveryTag, false,false);
}
else //正常消息处理
{
Console.WriteLine($"{DateTime.Now}来自死信队列获取的消息: {str.ToString()}，该消息被接受");
channel.BasicAck(ea.DeliveryTag, false);
}
};
}
}
await Task.文章来源地址19049.htmlDelay(5000);
}
}
}

通过模拟发送的消息加入跳过两个字会拒收这条消息，这样就会跳到设置的exchange.dlx交换机队列去，如果没有跳过那么这条消息就正常处理掉，消费确认。

超时不处理后我们通过新的消费服务DeadConsumerService来处理这异常的消费，比如回复库存，订单状态改为取消等等

public class DeadConsumerService:BackgroundService
{
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
Console.WriteLine("normal Rabbitmq消费端开始工作!");
while (!stoppingToken.IsCancellationRequested)
{
ConnectionFactory factory = new ConnectionFactory();
factory.HostName = "localhost";
factory.Port = 5672;

IConnection connection = factory.CreateConnection();
{
IModel channel = connection.CreateModel();
{
var queueName = "queue.dlx";
channel.ExchangeDeclare("exchange.dlx", ExchangeType.Direct, true);
channel.QueueDeclare("queue.dlx", true, false, false, null);

channel.QueueDeclare(queueName, true, false, false, null);
channel.QueueBind(queueName, "exchange.dlx", "");
//输入1，那如果接收一个消息，但是没有应答，则客户端不会收到下一个消息
channel.BasicQos(0, 1, false);
//在队列上定义一个消费者
var cOnsumer= new EventingBasicConsumer(channel);
channel.BasicConsume("queue.dlx", false, consumer);
consumer.Received += (ch, ea) =>
{
byte[] bytes = ea.Body.ToArray();
string str = Encoding.UTF8.GetString(bytes);
Console.WriteLine($"{DateTime.Now}超时未处理的消息: {str.ToString()}");
//回复确认
{
channel.BasicAck(ea.DeliveryTag, false);
}
};
文章来源地址19049.html}
}
await Task.Delay(5000);
www.yii666.com }
}
}

运行结果：

关于rabbitmq的死信队列和延时队列的介绍什么的这里不去贴baidu了，应用demo就这么多了，代码这里exercisebook/RabbitMQ.Test at main liuzhixin405/exercisebook (github.com) 。小面分享一个完整一点的例子。

exercisebook/cat.seckill/cat.seckill at main liuzhixin405/exercisebook (github.com)

感觉自己还是不合适写这些玩意儿，没有那么细心和耐心，有这时间真不如写写demo。

来源于：netcore下RabbitMQ队列、死信队列、延时队列及小应用

推荐阅读

php
C#学习教程：在Console中工作但在Windows窗体中不工作的异步代码分享

本文分享了一个关于在C#中使用异步代码的问题，作者在控制台中运行时代码正常工作，但在Windows窗体中却无法正常工作。作者尝试搜索局域网上的主机，但在窗体中计数器没有减少。文章提供了相关的代码和解决思路。 ... [详细]

蜡笔小新 2023-12-14 15:56:00
php
C# 7.0 新特性：基于Tuple的“多”返回值方法

本文介绍了C# 7.0中基于Tuple的“多”返回值方法的使用。通过对C# 6.0及更早版本的做法进行回顾，提出了问题：如何使一个方法可返回多个返回值。然后详细介绍了C# 7.0中使用Tuple的写法，并给出了示例代码。最后，总结了该新特性的优点。 ... [详细]

蜡笔小新 2023-12-13 19:55:20
php
后台获取视图对应的字符串

1.帮助类后台获取视图对应的字符串publicclassViewHelper{将View输出为字符串(注：不会执行对应的ac ... [详细]

蜡笔小新 2023-12-13 18:03:01
const
Java太阳系小游戏分析和源码详解

本文介绍了一个基于Java的太阳系小游戏的分析和源码详解。通过对面向对象的知识的学习和实践，作者实现了太阳系各行星绕太阳转的效果。文章详细介绍了游戏的设计思路和源码结构，包括工具类、常量、图片加载、面板等。通过这个小游戏的制作，读者可以巩固和应用所学的知识，如类的继承、方法的重载与重写、多态和封装等。 ... [详细]

蜡笔小新 2023-12-14 19:53:34
php
C#之数据集：DataSet对象的使用及相关方法详解

本文介绍了C#中数据集DataSet对象的使用及相关方法详解，包括DataSet对象的概述、与数据关系对象的互联、Rows集合和Columns集合的组成，以及DataSet对象常用的方法之一——Merge方法的使用。通过本文的阅读，读者可以了解到DataSet对象在C#中的重要性和使用方法。 ... [详细]

蜡笔小新 2023-12-14 12:09:13
get
JavaSE笔试题-接口、抽象类、多态等问题解答

本文解答了JavaSE笔试题中关于接口、抽象类、多态等问题。包括Math类的取整数方法、接口是否可继承、抽象类是否可实现接口、抽象类是否可继承具体类、抽象类中是否可以有静态main方法等问题。同时介绍了面向对象的特征，以及Java中实现多态的机制。 ... [详细]

蜡笔小新 2023-12-14 10:01:13
const
关于cuowu类的错误提示和使用AdjustmentListener的问题

本文讨论了一个关于cuowu类的问题，作者在使用cuowu类时遇到了错误提示和使用AdjustmentListener的问题。文章提供了16个解决方案，并给出了两个可能导致错误的原因。 ... [详细]

蜡笔小新 2023-12-13 22:09:56
io
Java高并发程序设计学习-线程安全的概念与synchronized

本文介绍了Java高并发程序设计中线程安全的概念与synchronized关键字的使用。通过一个计数器的例子，演示了多线程同时对变量进行累加操作时可能出现的问题。最终值会小于预期的原因是因为两个线程同时对变量进行写入时，其中一个线程的结果会覆盖另一个线程的结果。为了解决这个问题，可以使用synchronized关键字来保证线程安全。 ... [详细]

蜡笔小新 2023-12-13 15:43:03
io
Java中vector的使用详解

本文详细介绍了Java中vector的使用方法和相关知识，包括vector类的功能、构造方法和使用注意事项。通过使用vector类，可以方便地实现动态数组的功能，并且可以随意插入不同类型的对象，进行查找、插入和删除操作。这篇文章对于需要频繁进行查找、插入和删除操作的情况下，使用vector类是一个很好的选择。 ... [详细]

蜡笔小新 2023-12-13 14:14:39
php
猜字母游戏

猜字母游戏猜字母游戏——设计数据结构猜字母游戏——设计程序结构猜字母游戏——实现字母生成方法猜字母游戏——实现字母检测方法猜字母游戏——实现主方法1猜字母游戏——设计数据结构1.1 ... [详细]

蜡笔小新 2023-12-13 12:04:03
hash
[大整数乘法] java代码实现

本文介绍了使用java代码实现大整数乘法的过程，同时也涉及到大整数加法和大整数减法的计算方法。通过分治算法来提高计算效率，并对算法的时间复杂度进行了研究。详细代码实现请参考文章链接。 ... [详细]

蜡笔小新 2023-12-13 11:21:32
get
如何在dotnet桌面（Windows）应用程序中添加图标？

本文讨论了如何在dotnet桌面（Windows）应用程序中添加图标。作者提到可以使用dotnet命令行工具与resource.rc文件一起使用来为标准.NET核心应用程序添加图标。作者还介绍了在创建控制台应用程序时如何编辑projeto1.csproj文件来添加图标。 ... [详细]

蜡笔小新 2023-12-12 20:09:00
io
ScrollView嵌套Collectionview无痕衔接四向滚动，支持自定义TitleView

本文介绍了如何实现ScrollView嵌套Collectionview无痕衔接四向滚动，并支持自定义TitleView。通过使用MainScrollView作为最底层，headView作为上部分，TitleView作为中间部分，Collectionview作为下面部分，实现了滚动效果。同时还介绍了使用runtime拦截_notifyDidScroll方法来实现滚动代理的方法。具体实现代码可以在github地址中找到。 ... [详细]

蜡笔小新 2023-12-12 11:38:19
hash
模板引擎StringTemplate的使用方法和特点

本文介绍了模板引擎StringTemplate的使用方法和特点，包括强制Model和View的分离、Lazy-Evaluation、Recursive enable等。同时，还介绍了StringTemplate语法中的属性和普通字符的使用方法，并提供了向模板填充属性的示例代码。 ... [详细]

蜡笔小新 2023-12-11 21:45:03
const
vue cli 3.x移除console采坑记

本文记录了在vue cli 3.x中移除console的一些采坑经验，通过使用uglifyjs-webpack-plugin插件，在vue.config.js中进行相关配置，包括设置minimizer、UglifyJsPlugin和compress等参数，最终成功移除了console。同时，还包括了一些可能出现的报错情况和解决方法。 ... [详细]

蜡笔小新 2023-12-11 21:16:05

ouyan1985_998

这个家伙很懒，什么也没留下！

Tags | 热门标签

RankList | 热门文章