欢迎光临
我的个人博客网站

.Net Core 5.x Api开发笔记 — 消息队列RabbitMQ实现事件总线EventBus(二)

上一节说了事件总线

本节在消息队列中实现事件处理:.Net Core 5.x Api开发笔记 — 消息队列RabbitMQ实现事件总线EventBus(一)

既然是消息队列,就需要有生产者和消费者(订阅)

 1 public interface IMessageQueue  2 {  3     /// <summary>  4     /// 发布消息  5     /// </summary>  6     void Publish<T>(T message, string exchangeName = "default", string queueName = "") where T : class;  7   8     /// <summary>  9     /// 订阅消息 10     /// </summary> 11     void Consume<T>(Func<T, Task<bool>> func, string queueName = ""); 12 }

生产者端实现发布消息接口:

 1 public class RabbitMQPublishClient : IMessageQueue  2 {  3     private static ConcurrentDictionary<string, string> QueueDic = new ConcurrentDictionary<string, string>();  4     public RabbitMQPublishClient(){}  5       6     /// <summary>  7     /// 发布消息  8     /// </summary>  9     /// <typeparam name="T"></typeparam> 10     /// <param name="message">消息实体</param> 11     /// <param name="exchangeName">交换机名称(默认default)</param> 12     /// <param name="queueName">队列名称(默认类名)</param> 13     public void Publish<T>(T message, string exchangeName = "default", string queueName = "") where T : class 14     { 15         using (var conn = GetConnection()) 16         { 17             using (var channel = conn.CreateModel()) 18             { 19                 if (!QueueDic.ContainsKey(queueName) || QueueDic[queueName] != exchangeName) 20                 { 21                     CreateQueue(channel, exchangeName, queueName, true); 22                     QueueDic.TryAdd(queueName, exchangeName); 23                 } 24                 var props = channel.CreateBasicProperties(); 25                 props.Persistent = true;  //消息持久化 26                 props.DeliveryMode = 2;   //消息持久化 27                 props.CorrelationId = Guid.NewGuid().ToString(); 28  29                 string content = JsonConvert.SerializeObject(message); 30                 var body = Encoding.UTF8.GetBytes(content); 31                 channel.BasicPublish(exchange: exchangeName, routingKey: queueName, basicProperties: props, body: body); 32             } 33         } 34     } 35  36     /// <summary> 37     /// 创建队列,绑定到交换机 38     /// </summary> 39     private void CreateQueue(IModel channel, string exchangeName, string queueName, bool isDurable = true) 40     { 41         channel.ExchangeDeclare(exchangeName, ExchangeType.Direct, durable: isDurable, autoDelete: false, arguments: null); 42         channel.QueueDeclare(queueName, durable: isDurable, exclusive: false, autoDelete: false, arguments: null); 43         channel.QueueBind(queueName, exchangeName, routingKey: queueName); 44     } 45 }

消费者(订阅者)实现订阅接口

 1 public class RabbitMQConsumerClient : IMessageQueue  2 {  3     public RabbitMQConsumerClient(){}  4   5     /// <summary>  6     /// 订阅消息  7     /// </summary>  8     public void Consume<T>(Func<T, Task<bool>> func, string queueName = "")  9     { 10         string exchangeName = exchangeName.Equals("default") ? $"{customKey}.default" : $"{customKey}.{queueName}"; 11  12         var collection = GetConnection(); 13         var channel = collection.CreateModel(); 14         channel.ExchangeDeclare(exchangeName, "direct", durable: true, autoDelete: false, arguments: null); 15         channel.QueueDeclare(queue: queueName, durable: true, exclusive: false, autoDelete: false, arguments: null); 16         channel.QueueBind(queueName, exchangeName, routingKey: queueName, arguments: null); 17  18         //消费事件 19         var consumer = new EventingBasicConsumer(channel); 20         consumer.Received += async (sender, e) => 21         { 22             try 23             { 24                 string content = Encoding.UTF8.GetString(e.Body.ToArray()); 25                 T message = JsonConvert.DeserializeObject<T>(content); 26                 bool isSuccess = await func.Invoke(message);    //执行func委托  这里才是真正执行传入的事件处理程序 27                 if (isSuccess) 28                 { 29                     channel.BasicAck(e.DeliveryTag, false);     //手动确认消息消费成功 30                 } 31                 else 32                 { 33                    channel.BasicReject(e.DeliveryTag, true);    //手动打回队列,下次重新消费 34                 } 35             } 36             catch (Exception ex) 37             { 38                 channel.BasicAck(e.DeliveryTag, false); 39             } 40         }; 41  42         channel.BasicQos(0, 1, false);  //限制同时接收消息数量为1,true是将限制应用于channel级,false是将限制应用于consumer级 43         channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer);   //订阅消息   autoAck: false 表示手动确认消息 44     } 45 }

上边都是平时很常用的功能逻辑,接下来是才是重点!!!

1,事件总线有了

2,生产者订阅者也有了

接下来还需要做至少3件事:

1,创建事件处理程序,就是真正干活的

2,注册消费者、注册事件处理程序、绑定事件源和事件处理程序(在订阅端)

3,订阅端启动消费者订阅监控,并触发事件处理程序

4,发布消息测试

—————————————————————–

1,创建事件处理程序

 1 /// <summary>  2 /// 事件处理程序,真正干活的是我  3 /// 继承IEventHandler<T>接口,其中T必须用EventWithData<TData>类型,因为EventWithData<TData>继承了EventBase类  4 /// </summary>  5 public class Person_EventHandler : IEventHandler<EventWithData<Person>>  6 {  7     //最终干活的  8     public async Task HandleEvent(EventWithData<Person> eventWithData)  9     { 10         var data = eventWithData.Data; 11         if (data == null) 12         { 13             Console.WriteLine("么有数据呀!"); 14         } 15         else 16         { 17             try 18             { 19                 Console.WriteLine($"{DateTime.Now}-------" + JsonConvert.SerializeObject(data)); 20             } 21             catch (Exception ex) 22             { 23                 Console.WriteLine("异常:" + ex.Message); 24             } 25         } 26     } 27 }

2,注册消费者、注册事件处理程序、绑定事件源和事件处理程序(在订阅端),这一步在 Startup 中实现!!!

 1 public IServiceCollection ServiceCollection { get; private set; }  3 public void ConfigureServices(IServiceCollection services)  4 {  5     ServiceCollection = services;    //将services传递给ServiceCollection  6     services.AddControllersWithViews();  7   8     //注册消费者  9     services.AddSingleton<IMessageQueue, RabbitMQConsumerClient>(); 10     //注册事件处理程序,绑定事件源:IEventHandler<EventWithData<Person>>和事件处理程序:Person_EventHandler的关系 11     services.AddSingleton<Person_EventHandler>(); 12 } 13  14 public void Configure(IApplicationBuilder app, IWebHostEnvironment env) 15 { 16     //绑定事件源和事件处理程序的映射关系   17     //这里跟上边的绑定不是一回事,这里仅仅就是绑定了一个映射关系,等到后边事件触发的时候需要用到该处映射关系 18     EventBus.Default.RegisterHandler<EventWithData<Person>, Person_EventHandler>(); 19  20     //其它中间件。。 21     。。。 22  23     //启动消费者订阅监控入口 24     app.Consumer(app.ApplicationServices, ServiceCollection); 25 }

3,启动消费者订阅监控,并触发事件处理程序

 1 /// <summary>  2 /// 启动消费者订阅监控入口  3 /// </summary>  4 public static void Consumer(this IApplicationBuilder app, IServiceProvider provider, IServiceCollection services)  5 {  6     //获取实例对象   7     IMessageQueue queue = provider.GetService<IMessageQueue>();     8   9     //异步调用,这里接收的消息类型T为:EventWithData<Person>  也可以是其它自定义消息类型,没有限制 10     queue.Consume<EventWithData<Person>>(async message => 11     { 12         using (var currentContainer = services.BuildServiceProvider().CreateScope())   //使用系统内置服务容器 13         { 14             IocManager.Configure(currentContainer.ServiceProvider);                //将容器服务传递过去 15              16             var e = EventWithData<Person>.New(message.Data);      //这里要传递的消息类型必须使用 EventWithData<TData>类型初始化消息 17              18             await EventBus.Default.TriggerAsync(e);               //调用触发事件逻辑(需要提前绑定映射关系和注册容器) 19              20             return true; 21         } 22     }); 23 }

说明:

IocManager.Configure(currentContainer.ServiceProvider)  实际上就是声明了一个全局的 IServiceProvider 变量,然后将当前的ServiceProvider赋值给全局变量

await EventBus.Default.TriggerAsync(e)   执行的就是上一节事件总线中的 public async Task TriggerAsync<TEvent>(TEvent e)  方法

到这里 事件源和事件处理程序就一一对应上了。

接下来开始测试一下是否可用,先在发布的应用程序Startup中注册一下消息队列容器

1 services.AddSingleton<IMessageQueue, RabbitMQPublishClient>();

然后发布一条消息,注意:这里消息的类型EventWithData<Person>没有限制的,你可以不使用EventWithData<Person>类型的消息也行

 1 //发布消息  2 var message = EventWithData<Person>.New(  3     new Person()  4     {  5         UserId = 1,  6         UserName = "张三"  7     }  8 );  9  10 messageQueue.Publish(message);

最后测试,订阅者成功消费了刚刚生产的消息

.Net Core 5.x Api开发笔记 -- 消息队列RabbitMQ实现事件总线EventBus(二)

赞(0) 打赏
未经允许不得转载:张拓的天空 » .Net Core 5.x Api开发笔记 — 消息队列RabbitMQ实现事件总线EventBus(二)
分享到: 更多 (0)

评论 抢沙发

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址

专业的IT技术经验分享 更专业 更方便

联系我们本站主机

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

微信扫一扫打赏