NetCore基于EasyNetQ的高级API使用RabbitMq

一、消息队列 消息队列作为分布式系统中的重要组件,常用的有MSMQ,RabbitMq,Kafa,ActiveMQ,RocketMQ。至于各种消息队列的优缺点比较,在这里就不做扩展了,网上资源很多。 更多内容可参考 消息队列及常见消息队列介绍。我在这里选用的是RabbitMq。 官网地址:http://www.rabbitmq.com 安装和配置:Windows下RabbitMq安装及配置 二、RabbitMq简单介绍 RabbitMQ是一款基于AMQP(高级消息队列协议),由Erlang开发的开源消息队列组件。是一款优秀的消息队列组件,他由两部分组成:服务端和客户端,客户端支持多种语言的驱动,如:.Net、JAVA、 Erlang等。在RabbitMq中首先要弄清楚的概念是 交换机、队列、绑定。基本的消息通讯步骤就是首先定义ExChange,然后定义队列,然后绑定交换机和队列。 需要明确的一点儿是,发布者在发送消息是,并不是把消息直接发送到队列中,而是发送到Exchang,然后由交互机根据定义的消息匹配规则,在将消息发送到队列中。 Exchange有四种消息消息分发规则:direct,topic,fanout,header。headers 匹配 AMQP 消息的 header 而不是路由键,此外 headers 交换器和 direct 交换器完全一致,但性能差很多,目前几乎用不到了。 详细的概念介绍推荐查看:消息队列之RabbitMq 三、EasyNetQ使用 Easynetq是一个简单易用的Rabbitmq Net客户端。同时支持 NetFramework和NetCore。GitHub地址。它是针对RabbitMq Net客户端的进一步封装。关于EasyNetQ的简单使用推荐教程:EasyNetQ的介绍。 本文主要介绍基于EasyNeq的高级API的使用。EasyNetQ的作者在核心的IBus接口中尽量避免暴露AMQP中的交换机、队列、绑定这些概念,使用者即使不去了解这些概念,也能完成消息的发送接收。这相当简洁,但某些情况下,基于应用场景的需要,我们需要自定义交换机、队列、绑定这些信息,EasyNetQ允许你这么做,这些都是通过IAdvanceBus接口实现。 3.1 项目装备 这里为了演示,首先新建一个项目,包括一个发布者,两个接收者,一个公共的类库 安装EasyNetQ: NuGet>Install-Package EasyNetQ 3.2 简单封装 在Common项目里面是针对Easynetq的使用封装,主要目录如下 在RabbitMq文件夹下,是针对消息发送接收的简单封装。 首先来看下RabbitMqManage,主要的发送和订阅操作都在这个类中。其中ISend接口定义了发送消息的规范,SendMessageManage是ISend的实现。IMessageConsume接口定义订阅规范。 MesArg 和PushMsg分别是订阅和发送需用到的参数类。RabbitMQManage是暴露在外的操作类。 首先看发送的代码 复制代码 public enum SendEnum { 订阅模式 = 1, 推送模式 = 2, 主题路由模式 = 3 } public class PushMsg { /// /// 发送的数据 /// public object sendMsg { get; set; } /// /// 消息推送的模式 /// 现在支持:订阅模式,推送模式,主题路由模式 /// public SendEnum sendEnum { get; set; } /// /// 管道名称 /// public string exchangeName { get; set; } /// /// 路由名称 /// public string routeName { get; set; } } internal interface ISend { Task SendMsgAsync(PushMsg pushMsg, IBus bus); void SendMsg(PushMsg pushMsg, IBus bus); } internal class SendMessageMange : ISend { public async Task SendMsgAsync(PushMsg pushMsg, IBus bus) { //一对一推送 var message = new Message(pushMsg.sendMsg); IExchange ex = null; //判断推送模式 if (pushMsg.sendEnum == SendEnum.推送模式) { ex = bus.Advanced.ExchangeDeclare(pushMsg.exchangeName, ExchangeType.Direct); } if (pushMsg.sendEnum == SendEnum.订阅模式) { //广播订阅模式 ex = bus.Advanced.ExchangeDeclare(pushMsg.exchangeName, ExchangeType.Fanout); } if (pushMsg.sendEnum == SendEnum.主题路由模式) { //主题路由模式 ex = bus.Advanced.ExchangeDeclare(pushMsg.exchangeName, ExchangeType.Topic); } await bus.Advanced.PublishAsync(ex, pushMsg.routeName.ToSafeString(""), false, message) .ContinueWith(task => { if (!task.IsCompleted && task.IsFaulted)//消息投递失败 { //记录投递失败的消息信息 } }); } public void SendMsg(PushMsg pushMsg, IBus bus) { //一对一推送 var message = new Message(pushMsg.sendMsg); IExchange ex = null; //判断推送模式 if (pushMsg.sendEnum == SendEnum.推送模式) { ex = bus.Advanced.ExchangeDeclare(pushMsg.exchangeName, ExchangeType.Direct); } if (pushMsg.sendEnum == SendEnum.订阅模式) { //广播订阅模式 ex = bus.Advanced.ExchangeDeclare(pushMsg.exchangeName, ExchangeType.Fanout); } if (pushMsg.sendEnum == SendEnum.主题路由模式) { //主题路由模式 ex = bus.Advanced.ExchangeDeclare(pushMsg.exchangeName, ExchangeType.Topic); } bus.Advanced.Publish(ex, pushMsg.routeName.ToSafeString(""), false, message); } } 复制代码 在EasyNetQ中对于异步发送消息的时候,消息是否送达Broker只需要查看异步发送方法最终执行成功还是失败,成功就表示消息送达,如果失败可以将失败后的消息存入数据库中,然后用后台线程轮询 数据库表,将失败后的消息进行重新 发送。这种方式还可以进一步变成消息表,就是先将要发送的消息存入消息表中,然后后台线程轮询消息表来进行消息发送。一般这种方式被广泛用于分布式事务中, 将本地数据库操作和消息表写入放入同一个本地事务中,来保证消息发送和本地数据操作的同步成功,因为我的系统中,分布式事务的涉及很少,所以就没这样去做,只是简单的在异步发送的时候监控下 是否发送失败,然后针对失败的消息做一个重新发送的机制。这里,推荐大佬的NetCore分布式事务解决方案 CAP GitHub地址。 接着看一下消息订阅接收涉及的代码 复制代码 public class MesArgs { /// /// 消息推送的模式 /// 现在支持:订阅模式,推送模式,主题路由模式 /// public SendEnum sendEnum { get; set; } /// /// 管道名称 /// public string exchangeName { get; set; } /// /// 对列名称 /// public string rabbitQueeName { get; set; } /// /// 路由名称 /// public string routeName { get; set; } } public interface IMessageConsume { void Consume(string message); } 复制代码 在订阅中我定义了一个接口,最终业务代码中,所有的消息订阅类,都需要继续此接口 最后,我们来看下对外使用的操作类 复制代码 public class RabbitMQManage { private volatile static IBus bus = null; private static readonly object lockHelper = new object(); /// /// 创建服务总线 /// /// /// public static IBus CreateEventBus() { //获取RabbitMq的连接地址 //SystemJsonConfigManage 是我简单封装的一个json操作类,用于针对json文件的读写操作 var config = SystemJsonConfigManage.GetInstance().AppSettings["MeessageService"]; if (string.IsNullOrEmpty(config)) throw new Exception("消息地址未配置"); if (bus == null && !string.IsNullOrEmpty(config)) { lock (lockHelper) { if (bus == null) bus = RabbitHutch.CreateBus(config); } } return bus; } /// /// 释放服务总线 /// public static void DisposeBus() { bus?.Dispose(); } /// /// 消息同步投递 /// /// /// public static bool PushMessage(PushMsg pushMsg) { bool b = true; try { if (bus == null) CreateEventBus(); new SendMessageMange().SendMsg(pushMsg, bus); b = true; } catch (Exception ex) { b = false; } return b; } /// /// 消息异步投递 /// /// public static async Task PushMessageAsync(PushMsg pushMsg) { try { if (bus == null) CreateEventBus(); await new SendMessageMange().SendMsgAsync(pushMsg, bus); } catch (Exception ex) { throw ex; } } /// /// 消息订阅 /// public static void Subscribe(MesArgs args) where TConsum : IMessageConsume,new() { if (bus == null) CreateEventBus(); if (string.IsNullOrEmpty(args.exchangeName)) return; Expression> methodCall; IExchange ex = null; //判断推送模式 if (args.sendEnum == SendEnum.推送模式) { ex = bus.Advanced.ExchangeDeclare(args.exchangeName, ExchangeType.Direct); } if (args.sendEnum == SendEnum.订阅模式) { //广播订阅模式 ex = bus.Advanced.ExchangeDeclare(args.exchangeName, ExchangeType.Fanout); } if (args.sendEnum == SendEnum.主题路由模式) { //主题路由模式 ex = bus.Advanced.ExchangeDeclare(args.exchangeName, ExchangeType.Topic); } IQueue qu; if (string.IsNullOrEmpty(args.rabbitQueeName)) { qu = bus.Advanced.QueueDeclare(); } else qu = bus.Advanced.QueueDeclare(args.rabbitQueeName); bus.Advanced.Bind(ex, qu, args.routeName.ToSafeString("")); bus.Advanced.Consume(qu, (body, properties, info) => Task.Factory.StartNew(() => { try { lock (lockHelper) { var message = Encoding.UTF8.GetString(body); //处理消息 methodCall = job => job.Consume(message); methodCall.Compile()(new TConsum()); } } catch (Exception e) { throw e; } })); } } 复制代码 这里面主要封装了消息的发送和订阅,以及IBus单例的创建。在后续的消息发送和订阅主要就通过此处来实现。我们看到一开始的类目结构中还有一个RaExMessageHandleJob类,这个类就是一个后台 循环任务,用来监测数据库中是否保存了发送失败的消息,如果有,则将消息取出,尝试重新发送。在此就不做多的介绍,大家可以根据自己的实际需求来实现。 3.3 发布者 现在来看一下消息发布者的代码 主要的发送代码都在Send类中,其中appsettings.json里面配置了Rabbitmq的连接地址,TestDto只是一个为了方便演示的参数类。 下面看一下Program里面的代码 很简单的一个发送消息调用。 然后来看一下Send类中的代码 复制代码 public class Send { /// /// 发送消息 /// public static void SendMessage() { //需要注意一点儿,如果发送的时候,在该管道下找不到相匹配的队列框架将默认丢弃该消息 //推送模式 //推送模式下,需指定管道名称和路由键值名称 //消息只会被发送到和指定路由键值完全匹配的队列中 var directdto = new PushMsg() { sendMsg = new TestDto() { Var1 = "这是推送模式" }, exchangeName = "message.directdemo", routeName= "routekey", sendEnum =SendEnum.推送模式 }; //同步发送 ,返回true或fasle true 发送成功,消息已存储到Rabbitmq中,false表示发送失败 var b= RabbitMQManage.PushMessage(directdto); //异步发送,如果失败,失败的消息会被写入数据库,会有后台线程轮询数据库进行重新发送 //RabbitMQManage.PushMessageAsync(directlist); //订阅模式 //订阅模式只需要指定管道名称 //消息会被发送到该管道下的所有队列中 var fanoutdto = new PushMsg() { sendMsg = new TestDto() { Var1 = "这是订阅模式" }, exchangeName = "message.fanoutdemo", sendEnum = SendEnum.订阅模式 }; //同步发送 var fb = RabbitMQManage.PushMessage(fanoutdto); //异步发送 //RabbitMQManage.PushMessageAsync(fanoutdto); //主题路由模式 //路由模式下需指定 管道名称和路由值 //消息会被发送到该管道下,和路由值匹配的队列中去 var routedto = new PushMsg() { sendMsg = new TestDto() { Var1 = "这是主题路由模式1", }, exchangeName = "message.topicdemo", routeName="a.log", sendEnum=SendEnum.主题路由模式 }; var routedto2 = new PushMsg() { sendMsg = new TestDto() { Var1 = "这是主题路由模式2", }, exchangeName = "message.topicdemo", routeName = "a.log.a.b", sendEnum = SendEnum.主题路由模式 }; //同步发送 var rb = RabbitMQManage.PushMessage(routedto); var rb2 = RabbitMQManage.PushMessage(routedto2); //异步发送 //RabbitMQManage.PushMessageAsync(routedto); } } 复制代码 3.4 消费者 首先来看下消费者端的目录结构 其中appsettings.json中配置Rabbitmq的连接信息,Program中只是简单调用消息订阅 主要的消息订阅代码都在MessageManage文件夹下,MessageManService用于定义消息订阅类型 复制代码 public class MessageManService { public static void Subsribe() { Task.Run(() => { //概念 一个管道下面可以绑定多个队列。 //发送消息 是指将消息发送到管道中,然后由rabbitmq根据发送规则在将消息具体的转发到对应到管道下面的队列中 //消费消息 是指消费者(即服务)从管道下面的队列中获取消息 //同一个队列 可以有多个消费者(即不同的服务,都可以连接到同一个队列去获取消息) //但注意 当一个队列有多个消费者的时候,消息会被依次分发到不同的消费者中。比如第一条消息给第一个消费者,第二条消息给第二个消费者(框架内部有一个公平分发的机制) //推送模式时 需指定管道名称和路由值 //队列名称可自己指定 //注意 ,管道名称和路由名称一定要和发送方的管道名称和路由名称一致 //无论这个管道下面挂靠有多少个队列,只有路由名称和此处指定的路由名称完全一致的队列,才会收到这条消息。 var dirarg = new MesArgs() { sendEnum = SendEnum.推送模式, exchangeName = "message.directdemo", rabbitQueeName = "meesage.directmessagequene", routeName = "routekey" }; RabbitMQManage.Subscribe(dirarg); //订阅模式时需指定管道名称,并且管道名称要和发送方管道名称一致 //队列名称可自己指定 //所有这个管道下面的队列,都将收到该条消息 var fanoutrg = new MesArgs() { sendEnum = SendEnum.订阅模式
50000+
5万行代码练就真实本领
17年
创办于2008年老牌培训机构
1000+
合作企业
98%
就业率

联系我们

电话咨询

0532-85025005

扫码添加微信