.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(二)

上一篇:27、Android--EventBus


下一篇:vue系列---【组件通信、ref、is、slot插槽、mixin混入、脚手架使用animate】