.net平台的rabbitmq使用封装demo详解
前言
RabbitMq大家再熟悉不过,这篇文章主要针对rabbitmq学习后封装RabbitMQ.Client的一个分享。文章最后,我会把封装组件和demo奉上。
什么是rabbitMQ
RabbitMQ是一个由erlang开发的AMQP(Advanced Message Queue 高级消息队列协议 )的开源实现,
能够实现异步消息处理
RabbitMQ是一个消息代理:它接受和转发消息。
你可以把它想象成一个邮局:当你把你想要发布的邮件放在邮箱中时,你可以确定邮差先生最终将邮件发送给你的收件人。在这个比喻中,RabbitMQ是邮政信箱,邮局和邮递员。
RabbitMQ和邮局的主要区别在于它不处理纸张,而是接受,存储和转发二进制数据块
优点:异步消息处理
业务解耦(下订单操作:扣减库存、生成订单、发红包、发短信),
将下单操作主流程:扣减库存、生成订单
然后通过MQ消息队列完成通知,发红包、发短信
错峰流控 (通知量 消息量 订单量大的情况实现MQ消息队列机制,淡季情况下访问量会少)
灵活的路由(Flexible Routing)
在消息进入队列之前,通过 Exchange 来路由消息的。对于典型的路由功能,RabbitMQ 已经提供了一些内置的 Exchange 来实现。针对更复杂的路由功能,可以将多个 Exchange 绑定在一起,也通过插件机制实现自己的 Exchange 。
RabbitMQ网站端口号:15672
程序里面实现的端口为:5672
Rabbitmq的关键术语
1、绑定器(Binding):根据路由规则绑定Queue和Exchange。
2、路由键(Routing Key):Exchange根据关键字进行消息投递。
3、交换机(Exchange):指定消息按照路由规则进入指定队列
4、消息队列(Queue):消息的存储载体
5、生产者(Producer):消息发布者。
6、消费者(Consumer):消息接收者。
Rabbitmq的运作
从下图可以看出,发布者(Publisher)是把消息先发送到交换器(Exchange),再从交换器发送到指定队列(Queue),而先前已经声明交换器与队列绑定关系,最后消费者(Customer)通过订阅或者主动取指定队列消息进行消费。
那么刚刚提到的订阅和主动取可以理解成,推(被动),拉(主动)。
推,只要队列增加一条消息,就会通知空闲的消费者进行消费。(我不找你,就等你找我,观察者模式)
拉,不会通知消费者,而是由消费者主动轮循或者定时去取队列消息。(我需要才去找你)
使用场景我举个例子,假如有两套系统 订单系统和发货系统,从订单系统发起发货消息指令,为了及时发货,发货系统需要订阅队列,只要有指令就处理。
可是程序偶尔会出异常,例如网络或者DB超时了,把消息丢到失败队列,这个时候需要重发机制。但是我又不想while(IsPostSuccess == True),因为只要出异常了,会在某个时间段内都会有异常,这样的重试是没意义的。
这个时候不需要及时的去处理消息,有个JOB定时或者每隔几分钟(失败次数*间隔分钟)去取失败队列消息,进行重发。
Publish(发布)的封装
步骤:初始化链接->声明交换器->声明队列->换机器与队列绑定->发布消息。注意的是,我将Model存到了ConcurrentDictionary里面,因为声明与绑定是非常耗时的,其次,往重复的队列发送消息是不需要重新初始化的。
/// <summary> /// 交换器声明 /// </summary> /// <param name="iModel"></param> /// <param name="exchange">交换器</param> /// <param name="type">交换器类型: /// 1、Direct Exchange – 处理路由键。需要将一个队列绑定到交换机上,要求该消息与一个特定的路由键完全 /// 匹配。这是一个完整的匹配。如果一个队列绑定到该交换机上要求路由键 “dog”,则只有被标记为“dog”的 /// 消息才被转发,不会转发dog.puppy,也不会转发dog.guard,只会转发dog /// 2、Fanout Exchange – 不处理路由键。你只需要简单的将队列绑定到交换机上。一个发送到交换机的消息都 /// 会被转发到与该交换机绑定的所有队列上。很像子网广播,每台子网内的主机都获得了一份复制的消息。Fanout /// 交换机转发消息是最快的。 /// 3、Topic Exchange – 将路由键和某模式进行匹配。此时队列需要绑定要一个模式上。符号“#”匹配一个或多 /// 个词,符号“*”匹配不多不少一个词。因此“audit.#”能够匹配到“audit.irs.corporate”,但是“audit.*” /// 只会匹配到“audit.irs”。</param> /// <param name="durable">持久化</param> /// <param name="autoDelete">自动删除</param> /// <param name="arguments">参数</param> private static void ExchangeDeclare(IModel iModel, string exchange, string type = ExchangeType.Direct, bool durable = true, bool autoDelete = false, IDictionary<string, object> arguments = null) { exchange = exchange.IsNullOrWhiteSpace() ? "" : exchange.Trim(); iModel.ExchangeDeclare(exchange, type, durable, autoDelete, arguments); } /// <summary> /// 队列声明 /// </summary> /// <param name="channel"></param> /// <param name="queue">队列</param> /// <param name="durable">持久化</param> /// <param name="exclusive">排他队列,如果一个队列被声明为排他队列,该队列仅对首次声明它的连接可见, /// 并在连接断开时自动删除。这里需要注意三点:其一,排他队列是基于连接可见的,同一连接的不同信道是可 /// 以同时访问同一个连接创建的排他队列的。其二,“首次”,如果一个连接已经声明了一个排他队列,其他连 /// 接是不允许建立同名的排他队列的,这个与普通队列不同。其三,即使该队列是持久化的,一旦连接关闭或者 /// 客户端退出,该排他队列都会被自动删除的。这种队列适用于只限于一个客户端发送读取消息的应用场景。</param> /// <param name="autoDelete">自动删除</param> /// <param name="arguments">参数</param> private static void QueueDeclare(IModel channel, string queue, bool durable = true, bool exclusive = false, bool autoDelete = false, IDictionary<string, object> arguments = null) { queue = queue.IsNullOrWhiteSpace() ? "UndefinedQueueName" : queue.Trim(); channel.QueueDeclare(queue, durable, exclusive, autoDelete, arguments); } /// <summary> /// 获取Model /// </summary> /// <param name="exchange">交换机名称</param> /// <param name="queue">队列名称</param> /// <param name="routingKey"></param> /// <param name="isProperties">是否持久化</param> /// <returns></returns> private static IModel GetModel(string exchange, string queue, string routingKey, bool isProperties = false) { return ModelDic.GetOrAdd(queue, key => { var model = _conn.CreateModel(); ExchangeDeclare(model, exchange, ExchangeType.Fanout, isProperties); QueueDeclare(model, queue, isProperties); model.QueueBind(queue, exchange, routingKey); ModelDic[queue] = model; return model; }); } /// <summary> /// 发布消息 /// </summary> /// <param name="routingKey">路由键</param> /// <param name="body">队列信息</param> /// <param name="exchange">交换机名称</param> /// <param name="queue">队列名</param> /// <param name="isProperties">是否持久化</param> /// <returns></returns> public void Publish(string exchange, string queue, string routingKey, string body, bool isProperties = false) { var channel = GetModel(exchange, queue, routingKey, isProperties); try { channel.BasicPublish(exchange, routingKey, null, body.SerializeUtf8()); } catch (Exception ex) { throw ex.GetInnestException(); } }
下次是本机测试的发布速度截图:
4.2W/S属于稳定速度,把反序列化(ToJson)会稍微快一些。
Subscribe(订阅)的封装
发布的时候是申明了交换器和队列并绑定,然而订阅的时候只需要声明队列就可。从下面代码能看到,捕获到异常的时候,会把消息送到自定义的“死信队列”里,由另外的JOB进行定时重发,因此,finally是应答成功的。
/// <summary> /// 获取Model /// </summary> /// <param name="queue">队列名称</param> /// <param name="isProperties"></param> /// <returns></returns> private static IModel GetModel(string queue, bool isProperties = false) { return ModelDic.GetOrAdd(queue, value => { var model = _conn.CreateModel(); QueueDeclare(model, queue, isProperties); //每次消费的消息数 model.BasicQos(0, 1, false); ModelDic[queue] = model; return model; }); } /// <summary> /// 接收消息 /// </summary> /// <typeparam name="T"></typeparam> /// <param name="queue">队列名称</param> /// <param name="isProperties"></param> /// <param name="handler">消费处理</param> /// <param name="isDeadLetter"></param> public void Subscribe<T>(string queue, bool isProperties, Action<T> handler, bool isDeadLetter) where T : class { //队列声明 var channel = GetModel(queue, isProperties); var consumer = new EventingBasicConsumer(channel); consumer.Received += (model, ea) => { var body = ea.Body; var msgStr = body.DeserializeUtf8(); var msg = msgStr.FromJson<T>(); try { handler(msg); } catch (Exception ex) { ex.GetInnestException().WriteToFile("队列接收消息", "RabbitMq"); if (!isDeadLetter) PublishToDead<DeadLetterQueue>(queue, msgStr, ex); } finally { channel.BasicAck(ea.DeliveryTag, false); } }; channel.BasicConsume(queue, false, consumer); }
下次是本机测试的发布速度截图:
快的时候有1.9K/S,慢的时候也有1.7K/S
Pull(拉)的封装
直接上代码:
/// <summary> /// 获取消息 /// </summary> /// <typeparam name="T"></typeparam> /// <param name="exchange"></param> /// <param name="queue"></param> /// <param name="routingKey"></param> /// <param name="handler">消费处理</param> private void Poll<T>(string exchange, string queue, string routingKey, Action<T> handler) where T : class { var channel = GetModel(exchange, queue, routingKey); var result = channel.BasicGet(queue, false); if (result.IsNull()) return; var msg = result.Body.DeserializeUtf8().FromJson<T>(); try { handler(msg); } catch (Exception ex) { ex.GetInnestException().WriteToFile("队列接收消息", "RabbitMq"); } finally { channel.BasicAck(result.DeliveryTag, false); } }
快的时候有1.8K/s,稳定是1.5K/S
Rpc(远程调用)的封装
首先说明下,RabbitMq只是提供了这个RPC的功能,但是并不是真正的RPC,为什么这么说:
1、传统Rpc隐藏了调用细节,像调用本地方法一样传参、抛出异常
2、RabbitMq的Rpc是基于消息的,消费者消费后,通过新队列返回响应结果。
/// <summary> /// RPC客户端 /// </summary> /// <param name="exchange"></param> /// <param name="queue"></param> /// <param name="routingKey"></param> /// <param name="body"></param> /// <param name="isProperties"></param> /// <returns></returns> public string RpcClient(string exchange, string queue, string routingKey, string body, bool isProperties = false) { var channel = GetModel(exchange, queue, routingKey, isProperties); var consumer = new QueueingBasicConsumer(channel); channel.BasicConsume(queue, true, consumer); try { var correlationId = Guid.NewGuid().ToString(); var basicProperties = channel.CreateBasicProperties(); basicProperties.ReplyTo = queue; basicProperties.CorrelationId = correlationId; channel.BasicPublish(exchange, routingKey, basicProperties, body.SerializeUtf8()); var sw = Stopwatch.StartNew(); while (true) { var ea = consumer.Queue.Dequeue(); if (ea.BasicProperties.CorrelationId == correlationId) { return ea.Body.DeserializeUtf8(); } if (sw.ElapsedMilliseconds > 30000) throw new Exception("等待响应超时"); } } catch (Exception ex) { throw ex.GetInnestException(); } } /// <summary> /// RPC服务端 /// </summary> /// <typeparam name="T"></typeparam> /// <param name="exchange"></param> /// <param name="queue"></param> /// <param name="isProperties"></param> /// <param name="handler"></param> /// <param name="isDeadLetter"></param> public void RpcService<T>(string exchange, string queue, bool isProperties, Func<T, T> handler, bool isDeadLetter) { //队列声明 var channel = GetModel(queue, isProperties); var consumer = new EventingBasicConsumer(channel); consumer.Received += (model, ea) => { var body = ea.Body; var msgStr = body.DeserializeUtf8(); var msg = msgStr.FromJson<T>(); var props = ea.BasicProperties; var replyProps = channel.CreateBasicProperties(); replyProps.CorrelationId = props.CorrelationId; try { msg = handler(msg); } catch (Exception ex) { ex.GetInnestException().WriteToFile("队列接收消息", "RabbitMq"); } finally { channel.BasicPublish(exchange, props.ReplyTo, replyProps, msg.ToJson().SerializeUtf8()); channel.BasicAck(ea.DeliveryTag, false); } }; channel.BasicConsume(queue, false, consumer); }
可以用,但不建议去用。可以考虑其他的RPC框架。grpc、thrift等。
结尾
本篇文章,没有过多的写RabbitMq的知识点,因为园子的学习笔记实在太多了。下面把我的代码奉上 https://github.com/SkyChenSky/Sikiro.Mq.Rabbit。如果有发现写得不对的地方麻烦在评论指出,我会及时修改以免误导别人。
到此这篇关于.net平台的rabbitmq使用封装的文章就介绍到这了,更多相关.net使用rabbitmq内容请搜索猪先飞以前的文章或继续浏览下面的相关文章希望大家以后多多支持猪先飞!
相关文章
- 这篇文章主要为大家详细介绍了ASP.NET购物车的实现过程,文中示例代码介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们可以参考一下...2021-09-22
- 这篇文章主要介绍了.NET Core下使用Kafka的方法步骤,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧...2021-09-22
- 在开发过程中,使用Visual Studio的断点调试功能可以很方便帮我们调试发现程序存在的错误,同样Visual Studio也支持对SQL Server里面的存储过程进行调试,下面就让我们看看具体的调试方法。...2021-09-22
- 这篇文章主要介绍了Win10 IIS 安装及.net 4.5及Win10安装IIS并配置ASP.NET 4.0的方法,本文给大家介绍的非常详细,具有一定的参考借鉴价值,需要的朋友可以参考下...2021-09-22
- 这篇文章主要介绍了详解.NET Core 3.0 里新的JSON API,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧...2021-09-22
- 这篇文章主要介绍了.net数据库操作框架SqlSugar的简单入门,帮助大家更好的理解和学习使用.net技术,感兴趣的朋友可以了解下...2021-09-22
ASP.NET Core根据环境变量支持多个 appsettings.json配置文件
这篇文章主要介绍了ASP.NET Core根据环境变量支持多个 appsettings.json配置文件,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧...2021-09-22- 这篇文章主要介绍了记一次EFCore类型转换错误及解决方案,帮助大家更好的理解和学习使用asp.net core,感兴趣的朋友可以了解下...2021-09-22
- ZXing是一个开放源码的,用Java实现的多种格式的1D/2D条码图像处理库,它包含了联系到其他语言的端口。这篇文章主要给大家介绍了.NET C#利用ZXing生成、识别二维码/条形码的方法,文中给出了详细的示例代码,有需要的朋友们可以参考借鉴。...2020-06-25
详解ASP.NET Core 中基于工厂的中间件激活的实现方法
这篇文章主要介绍了ASP.NET Core 中基于工厂的中间件激活的实现方法,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下...2021-09-22C#使用RabbitMq队列(Sample,Work,Fanout,Direct等模式的简单使用)
这篇文章主要介绍了C#使用RabbitMq队列(Sample,Work,Fanout,Direct等模式的简单使用),本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下...2020-12-08- 这篇文章主要为大家详细介绍了C#操作RabbitMQ的完整实例,具有一定的参考价值,感兴趣的小伙伴们可以参考一下...2020-06-25
- 这篇文章主要介绍了C#使用Ado.Net更新和添加数据到Excel表格的方法,较为详细的分析了OLEDB的原理与使用技巧,可实现较为方便的操作Excel数据,需要的朋友可以参考下...2020-06-25
asp.net通过消息队列处理高并发请求(以抢小米手机为例)
这篇文章主要介绍了asp.net通过消息队列处理高并发请求(以抢小米手机为例),文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧...2021-09-22ASP.NET单选按钮控件RadioButton常用属性和方法介绍
RadioButton又称单选按钮,其在工具箱中的图标为 ,单选按钮通常成组出现,用于提供两个或多个互斥选项,即在一组单选钮中只能选择一个...2021-09-22ASP.NET 2.0中的数据操作:使用两个DropDownList过滤的主/从报表
在前面的指南中我们研究了如何显示一个简单的主/从报表, 该报表使用DropDownList和GridView控件, DropDownList填充类别,GridView显示选定类别的产品. 这类报表用于显示具有...2016-05-19详解.NET Core 使用HttpClient SSL请求出错的解决办法
这篇文章主要介绍了.NET Core 使用HttpClient SSL请求出错的解决办法,小编觉得挺不错的,现在分享给大家,也给大家做个参考。一起跟随小编过来看看吧...2021-09-22- 这篇文章主要介绍了Python调用.NET库的方法步骤,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧...2020-05-09
ASP.NET中iframe框架点击左边页面链接 右边显示链接页面内容
这篇文章主要介绍了ASP.NET中iframe框架点击左边页面链接,右边显示链接页面内容的实现代码,感兴趣的小伙伴们可以参考一下...2021-09-22- ASP.NET Web API具有与ASP.NET MVC类似的编程方式,ASP.NET Web API不仅仅具有一个完全独立的消息处理管道,而且这个管道比为ASP.NET MVC设计的管道更为复杂,功能也更为强大。下面创建一个简单的Web API项目,需要的朋友可以参考下...2021-09-22