基于事件总线EventBus实现邮件推送功能

吉普

       有时候，有人给我的网站留了言，但是我必须要打开我的网站（https://www.xiandanplay.com/）才知道，所以我便决定给网站增加一个邮件推送的功能，好让我第一时间知道。于是乎，按照我自己的思路，同时为了去学习了解rabbitmq以及EventBus概念，我便设计了一套邮件推送的功能，这里分享出来，可能方案不是很好，大家不喜勿喷。
什么是事件总线   
    事件总线是对发布-订阅模式的一种实现。它是一种集中式事件处理机制，允许不同的组件之间进行彼此通信而又不需要相互依赖，达到一种解耦的目的。
    关于这个概念，网上有很多讲解的，这里我推荐一个讲的比较好的（事件总线知多少）
什么是RabbitMQ
   RabbitMQ这个就不用说了，想必到家都知道。
粗糙流程图

简单来解释就是：
      1、定义一个事件抽象类

1. public abstract class EventData
2.     {
3.         /// <summary>
4.         /// 唯一标识
5.         /// </summary>
6.         public string Unique { get; set; }
7.         /// <summary>
8.         /// 是否成功
9.         /// </summary>
10.         public bool Success { get; set; }
11.         /// <summary>
12.         /// 结果
13.         /// </summary>
14.         public string Result { get; set; }
15.     }

复制代码

　2、定义一个事件处理抽象类，以及对应的一个队列消息执行的一个记录

1. public abstract class EventHandler<T> where T : EventData
2.     {
3.         public async Task Handler(T eventData)
4.         {
5.             await BeginHandler(eventData.Unique);
6.             eventData = await ProcessingHandler(eventData);
7.             if (eventData.Success)
8.                 await FinishHandler(eventData);
9.         }
10.         /// <summary>
11.         ///  开始处理
12.         /// </summary>
13.         /// <param name="unique"></param>
14.         /// <returns></returns>
15.         protected abstract Task BeginHandler(string unique);
16.         /// <summary>
17.         /// 处理中
18.         /// </summary>
19.         /// <param name="eventData"></param>
20.         /// <returns></returns>
21.         protected abstract Task<T> ProcessingHandler(T eventData);
22.         /// <summary>
23.         /// 处理完成
24.         /// </summary>
25.         /// <param name="eventData"></param>
26.         /// <returns></returns>
27.         protected abstract Task FinishHandler(T eventData);
28.     }
29.    
30.    [Table("Sys_TaskRecord")]
31.     public class TaskRecord : Entity<long>
32.     {
33.         /// <summary>
34.         /// 任务类型
35.         /// </summary>
36.         public TaskRecordType TaskType { get; set; }
37.         /// <summary>
38.         /// 任务状态
39.         /// </summary>
40.         public int TaskStatu { get; set; }
41.         /// <summary>
42.         /// 任务值
43.         /// </summary>
44.         public string TaskValue { get; set; }
45.         /// <summary>
46.         /// 任务结果
47.         /// </summary>
48.         public string TaskResult { get; set; }
49.         /// <summary>
50.         /// 任务开始时间
51.         /// </summary>
52.         public DateTime TaskStartTime { get; set; }
53.         /// <summary>
54.         /// 任务完成时间
55.         /// </summary>
56.         public DateTime? TaskFinishTime { get; set; }
57.         /// <summary>
58.         /// 任务最后更新时间
59.         /// </summary>
60.         public DateTime? LastUpdateTime { get; set; }
61.         /// <summary>
62.         /// 任务名称
63.         /// </summary>
64.         public string TaskName { get; set; }
65.         /// <summary>
66.         /// 附加数据
67.         /// </summary>
68.         public string AdditionalData { get; set; }
69.     }

复制代码

　　 3、定义一个邮件事件消息类，继承自EventData，以及一个邮件处理的Hanler继承自EventHandler

1. public class EmailEventData:EventData
2.     {
3.         /// <summary>
4.         /// 邮件内容
5.         /// </summary>
6.         public string Body { get; set; }
7.         /// <summary>
8.         /// 接收者
9.         /// </summary>
10.         public string Reciver { get; set; }
11.     }
13. public class CreateEmailHandler<T> : Core.EventBus.EventHandler<T> where T : EventData
14.     {
15.         private IEmailService emailService;
16.         private IUnitOfWork unitOfWork;
17.         private ITaskRecordService taskRecordService;
18.         public CreateEmailHandler(IEmailService emailService, IUnitOfWork unitOfWork, ITaskRecordService taskRecordService)
19.         {
20.             this.emailService = emailService;
21.             this.unitOfWork = unitOfWork;
22.             this.taskRecordService = taskRecordService;
23.         }
24.         protected override async Task BeginHandler(string unique)
25.         {
26.             await taskRecordService.UpdateRecordStatu(Convert.ToInt64(unique), (int)MqMessageStatu.Processing);
27.             await unitOfWork.CommitAsync();
28.         }
30.         protected override async Task<T> ProcessingHandler(T eventData)
31.         {
32.             try
33.             {
34.                 EmailEventData emailEventData = eventData as EmailEventData;
35.                 await emailService.SendEmail(emailEventData.Reciver, emailEventData.Reciver, emailEventData.Body, "[闲蛋]收到一条留言");
36.                 eventData.Success = true;
37.             }
38.             catch (Exception ex)
39.             {
40.                 await taskRecordService.UpdateRecordFailStatu(Convert.ToInt64(eventData.Unique), (int)MqMessageStatu.Fail,ex.Message);
41.                 await unitOfWork.CommitAsync();
42.                 eventData.Success = false;
43.             }
44.             return eventData;
46.         }
48.         protected override async Task FinishHandler(T eventData)
49.         {
50.             await taskRecordService.UpdateRecordSuccessStatu(Convert.ToInt64(eventData.Unique), (int)MqMessageStatu.Finish,"");
51.             await unitOfWork.CommitAsync();
52.         }

复制代码

　　 4、接着就是如何把事件消息和事件Hanler关联起来，那么我这里思路就是把EmailEventData的类型和CreateEmailHandler的类型先注册到字典里面，这样我就可以根据EmailEventData找到对应的处理程序了，找类型还不够，如何创建实例呢，这里就还需要把CreateEmailHandler注册到DI容器里面，这样就可以根据容器获取对象了，如下

1.   public void AddSub<T, TH>()
2.              where T : EventData
3.              where TH : EventHandler<T>
4.         {
5.             Type eventDataType = typeof(T);
6.             Type handlerType = typeof(TH);
7.             if (!eventhandlers.ContainsKey(typeof(T)))
8.                 eventhandlers.TryAdd(eventDataType, handlerType);
9.             _serviceDescriptors.AddScoped(handlerType);
10.         }
11. -------------------------------------------------------------------------------------------------------------------
12. public Type FindEventType(string eventName)
13.         {
14.             if (!eventTypes.ContainsKey(eventName))
15.                 throw new ArgumentException(string.Format("eventTypes不存在类名{0}的key", eventName));
16.             return eventTypes[eventName];
17.         }
18. ------------------------------------------------------------------------------------------------------------------------------------------------------------
19.   public object FindHandlerType(Type eventDataType)
20.         {
21.             if (!eventhandlers.ContainsKey(eventDataType))
22.                 throw new ArgumentException(string.Format("eventhandlers不存在类型{0}的key", eventDataType.FullName));
23.             var obj = _buildServiceProvider(_serviceDescriptors).GetService(eventhandlers[eventDataType]);
24.             return obj;
25.         }
26. ----------------------------------------------------------------------------------------------------------------------------------
27. private static IServiceCollection AddEventBusService(this IServiceCollection services)
28.         {
29.             string exchangeName = ConfigureProvider.configuration.GetSection("EventBusOption:ExchangeName").Value;
30.             services.AddEventBus(Assembly.Load("XianDan.Application").GetTypes())
31.                 .AddSubscribe<EmailEventData, CreateEmailHandler<EmailEventData>>(exchangeName, ExchangeType.Direct, BizKey.EmailQueueName);
32.             return services;
33.         }

复制代码

　　 5、发送消息，这里代码简单，就是简单的发送消息，这里用eventData.GetType().Name作为消息的RoutingKey，这样消费这就可以根据这个key调用FindEventType，然后找到对应的处理程序了

1. using (IModel channel = connection.CreateModel())
2. {
3.      string routeKey = eventData.GetType().Name;
4.      string message = JsonConvert.SerializeObject(eventData);
5.      byte[] body = Encoding.UTF8.GetBytes(message);
6.      channel.ExchangeDeclare(exchangeName, exchangeType, true, false, null);
7.      channel.QueueDeclare(queueName, true, false, false, null);
8.      channel.BasicPublish(exchangeName, routeKey, null, body);
9. }

复制代码

　　
6、订阅消息，核心的是这一段
  Type eventType = _eventBusManager.FindEventType(eventName);
  var eventData = (T)JsonConvert.DeserializeObject(body, eventType);
  EventHandler eventHandler = _eventBusManager.FindHandlerType(eventType)  as       EventHandler;

1. public void Subscribe<T, TH>(string exchangeName, string exchangeType, string queueName)
2.             where T : EventData
3.             where TH : EventHandler<T>
4.         {
5.             try
6.             {
7.                 _eventBusManager.AddSub<T, TH>();
8.                 IModel channel = connection.CreateModel();
9.                 channel.QueueDeclare(queueName, true, false, false, null);
10.                 channel.ExchangeDeclare(exchangeName, exchangeType, true, false, null);
11.                 channel.QueueBind(queueName, exchangeName, typeof(T).Name, null);
12.                 var consumer = new EventingBasicConsumer(channel);
13.                 consumer.Received += async (model, ea) =>
14.                 {
15.                     string eventName = ea.RoutingKey;
16.                     byte[] resp = ea.Body.ToArray();
17.                     string body = Encoding.UTF8.GetString(resp);
18.                     try
19.                     {
20.                         Type eventType = _eventBusManager.FindEventType(eventName);
21.                         var eventData = (T)JsonConvert.DeserializeObject(body, eventType);
22.                         EventHandler<T> eventHandler = _eventBusManager.FindHandlerType(eventType) as EventHandler<T>;
23.                         await eventHandler.Handler(eventData);
24.                     }
25.                     catch (Exception ex)
26.                     {
27.                         LogUtils.LogError(ex, "EventBusRabbitMQ", ex.Message);
28.                     }
29.                     finally
30.                     {
31.                         channel.BasicAck(ea.DeliveryTag, false);
32.                     }
33.                   
35.                 };
36.                 channel.BasicConsume(queueName, autoAck: false, consumer: consumer);
37.             }
38.             catch (Exception ex)
39.             {
40.                 LogUtils.LogError(ex, "EventBusRabbitMQ.Subscribe", ex.Message);
41.             }
43.         }

复制代码

　　  注意，这里我使用的时候有个小坑，就是最开始是用using包裹这个IModel channel = connection.CreateModel();导致最后程序启动后无法收到消息，然后去rabbitmq的管理界面发现没有channel连接，队列也没有消费者，最后发现可能是using执行完后就释放掉了，把using去掉就好了。
   好了，到此，我的思路大概讲完了，现在我的网站留言也可以收到邮件了，那么多测试邮件，哈哈哈哈哈

 
  大家感兴趣的话可以去我的网站（https://www.xiandanplay.com/）踩一踩，互加友链也可以的，谢谢大家，不喜勿喷喽！

来源:https://www.cnblogs.com/MrHanBlog/p/18381572
免责声明：由于采集信息均来自互联网，如果侵犯了您的权益，请联系我们【E-Mail:cb@itdo.tech】 我们会及时删除侵权内容，谢谢合作！