热门标签 | HotTags
当前位置:  开发笔记 > 编程语言 > 正文

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

原标题: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代码如下:

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 = "rbTest202301";
channel.QueueDeclare(queueName,
true, false, false, null);
//输入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(
"队列消息:" + str.ToString());
//回复确认
channel.BasicAck(ea.DeliveryTag, false);
};
}
}
await Task.Delay(5000);
}
}
}

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队列、死信队列、延时队列及小应用


推荐阅读
  • 并发编程 12—— 任务取消与关闭 之 shutdownNow 的局限性
    Java并发编程实践目录并发编程01——ThreadLocal并发编程02——ConcurrentHashMap并发编程03——阻塞队列和生产者-消费者模式并发编程04——闭锁Co ... [详细]
  • 2023年京东Android面试真题解析与经验分享
    本文由一位拥有6年Android开发经验的工程师撰写,详细解析了京东面试中常见的技术问题。涵盖引用传递、Handler机制、ListView优化、多线程控制及ANR处理等核心知识点。 ... [详细]
  • 本文介绍了Java并发库中的阻塞队列(BlockingQueue)及其典型应用场景。通过具体实例,展示了如何利用LinkedBlockingQueue实现线程间高效、安全的数据传递,并结合线程池和原子类优化性能。 ... [详细]
  • 微软Exchange服务器遭遇2022年版“千年虫”漏洞
    微软Exchange服务器在新年伊始遭遇了一个类似于‘千年虫’的日期处理漏洞,导致邮件传输受阻。该问题主要影响配置了FIP-FS恶意软件引擎的Exchange 2016和2019版本。 ... [详细]
  • 深入解析 Android IPC 中的 Messenger 机制
    本文详细介绍了 Android 中基于消息传递的进程间通信(IPC)机制——Messenger。通过实例和源码分析,帮助开发者更好地理解和使用这一高效的通信工具。 ... [详细]
  • Explore a common issue encountered when implementing an OAuth 1.0a API, specifically the inability to encode null objects and how to resolve it. ... [详细]
  • 1:有如下一段程序:packagea.b.c;publicclassTest{privatestaticinti0;publicintgetNext(){return ... [详细]
  • 使用 Azure Service Principal 和 Microsoft Graph API 获取 AAD 用户列表
    本文介绍了一段通用代码示例,该代码不仅能够操作 Azure Active Directory (AAD),还可以通过 Azure Service Principal 的授权访问和管理 Azure 订阅资源。Azure 的架构可以分为两个层级:AAD 和 Subscription。 ... [详细]
  • 本文详细介绍了Java编程语言中的核心概念和常见面试问题,包括集合类、数据结构、线程处理、Java虚拟机(JVM)、HTTP协议以及Git操作等方面的内容。通过深入分析每个主题,帮助读者更好地理解Java的关键特性和最佳实践。 ... [详细]
  • ServiceStack与Swagger的无缝集成指南
    本文详细介绍了如何在ServiceStack项目中集成Swagger,以实现API文档的自动生成和在线测试。通过本指南,您将了解从配置到部署的完整流程,并掌握如何优化API接口的开发和维护。 ... [详细]
  • 本文提供了使用Java实现Bellman-Ford算法解决POJ 3259问题的代码示例,详细解释了如何通过该算法检测负权环来判断时间旅行的可能性。 ... [详细]
  • 深入解析Redis内存对象模型
    本文详细介绍了Redis内存对象模型的关键知识点,包括内存统计、内存分配、数据存储细节及优化策略。通过实际案例和专业分析,帮助读者全面理解Redis内存管理机制。 ... [详细]
  • 深入解析for与foreach遍历集合时的性能差异
    本文将详细探讨for循环和foreach(迭代器)在遍历集合时的性能差异,并通过实际代码示例和源码分析,帮助读者理解这两种遍历方式的不同之处。文章内容丰富且专业,旨在为编程爱好者提供有价值的参考。 ... [详细]
  • 深入解析RDMA中的队列对(Queue Pair)
    本文将详细探讨RDMA架构中的关键组件——队列对(Queue Pair,简称QP),包括其基本概念、硬件与软件实现、QPC的作用、QPN的分配机制以及用户接口和状态机。通过这些内容,读者可以更全面地理解QP在RDMA通信中的重要性和工作原理。 ... [详细]
  • 深入理解Java多线程并发处理:基础与实践
    本文探讨了Java中的多线程并发处理机制,从基本概念到实际应用,帮助读者全面理解并掌握多线程编程技巧。通过实例解析和理论阐述,确保初学者也能轻松入门。 ... [详细]
author-avatar
ouyan1985_998
这个家伙很懒,什么也没留下!
PHP1.CN | 中国最专业的PHP中文社区 | DevBox开发工具箱 | json解析格式化 |PHP资讯 | PHP教程 | 数据库技术 | 服务器技术 | 前端开发技术 | PHP框架 | 开发工具 | 在线工具
Copyright © 1998 - 2020 PHP1.CN. All Rights Reserved | 京公网安备 11010802041100号 | 京ICP备19059560号-4 | PHP1.CN 第一PHP社区 版权所有