.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 /// <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 }

 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 }

 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 }

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

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