登录/ 注册
 找回密码
 立即注册
关闭

CSDN热搜

程序园 精品问答 技术交流 资源下载
返回列表 发新帖
首页 › 业界区 › 安全 ›使用 .NET 实现 EventBus（事件总线）设计模式 ...

使用 .NET 实现 EventBus（事件总线）设计模式

胆饬 6 天前
EventBus（事件总线）是一种设计模式，它允许应用程序的不同部分之间进行松散耦合的通信。发布者（Publisher）将事件发布到总线，而订阅者（Subscriber）监听总线上的特定事件类型，并在事件发生时执行相应的操作。发布者和订阅者之间不需要直接引用。
核心概念:

  • 事件 (Event): 表示系统中发生的事情，通常是一个简单的 POCO (Plain Old CLR Object) 类。
  • 发布者 (Publisher): 创建并发送事件到 EventBus 的组件。
  • 订阅者 (Subscriber): 注册对特定类型事件感兴趣，并在事件发布时接收通知的组件。
  • 事件总线 (EventBus): 负责管理订阅和将事件路由到相应订阅者的中心枢纽。
设计思路:

  • 接口 (IEventBus): 定义 EventBus 的核心功能：订阅、取消订阅、发布。
  • 实现 (EventBus):

    • 使用一个数据结构（如 Dictionary）来存储事件类型与对应的订阅者列表。
    • 键 (Key) 是事件的 Type。
    • 值 (Value) 是一个委托列表（如 List> 或 List，后者需要进行类型转换）。
    • 需要考虑线程安全，因为订阅、取消订阅和发布可能在不同线程上发生。使用 lock 或 ConcurrentDictionary 等机制。

  • 泛型方法: 使用泛型使订阅和发布类型安全。
1. 定义事件 (Event)
  1. // 事件可以是任何类，通常包含事件相关的数据
  2. public class OrderCreatedEvent
  3. {
  4.     public int OrderId { get; }
  5.     public DateTime Timestamp { get; }
  7.     public OrderCreatedEvent(int orderId)
  8.     {
  9.         OrderId = orderId;
  10.         Timestamp = DateTime.UtcNow;
  11.     }
  13.     public override string ToString()
  14.     {
  15.         return $"订单已创建: ID={OrderId}, 时间={Timestamp}";
  16.     }
  17. }
  19. public class UserLoggedInEvent
  20. {
  21.     public string Username { get; }
  22.     public DateTime LoginTime { get; }
  24.     public UserLoggedInEvent(string username)
  25.     {
  26.         Username = username;
  27.         LoginTime = DateTime.UtcNow;
  28.     }
  30.     public override string ToString()
  31.     {
  32.         return $"用户已登录: 用户名={Username}, 时间={LoginTime}";
  33.     }
  34. }
复制代码
2. 定义 EventBus 接口 (IEventBus)
  1. using System;
  3. public interface IEventBus
  4. {
  5.     /// <summary>
  6.     /// 订阅指定类型的事件
  7.     /// </summary>
  8.     /// <typeparam name="TEvent">事件类型</typeparam>
  9.     /// <param name="handler">事件处理委托</param>
  10.     void Subscribe<TEvent>(Action<TEvent> handler) where TEvent : class;
  12.     /// <summary>
  13.     /// 取消订阅指定类型的事件
  14.     /// </summary>
  15.     /// <typeparam name="TEvent">事件类型</typeparam>
  16.     /// <param name="handler">之前订阅时使用的同一个事件处理委托实例</param>
  17.     void Unsubscribe<TEvent>(Action<TEvent> handler) where TEvent : class;
  19.     /// <summary>
  20.     /// 发布一个事件，所有订阅了该事件类型的处理程序都将被调用
  21.     /// </summary>
  22.     /// <typeparam name="TEvent">事件类型</typeparam>
  23.     /// <param name="event">要发布的事件实例</param>
  24.     void Publish<TEvent>(TEvent @event) where TEvent : class;
  25. }
复制代码
3. 实现 EventBus (EventBus)
  1. using System;
  2. using System.Collections.Generic;
  3. using System.Linq;
  4. using System.Threading; // 用于演示或简单场景，实际可能用 Task
  6. public class EventBus : IEventBus
  7. {
  8.     // 存储订阅信息的字典
  9.     // Key: 事件类型 (Type)
  10.     // Value: 处理该事件类型的委托列表 (List of Action<TEvent>)，这里用 List<object> 存储以便处理不同泛型类型
  11.     private readonly Dictionary<Type, List<object>> _subscribers;
  12.     // 用于确保线程安全的锁对象
  13.     private readonly object _lock = new object();
  15.     public EventBus()
  16.     {
  17.         _subscribers = new Dictionary<Type, List<object>>();
  18.     }
  20.     public void Subscribe<TEvent>(Action<TEvent> handler) where TEvent : class
  21.     {
  22.         // 获取事件类型
  23.         Type eventType = typeof(TEvent);
  25.         lock (_lock)
  26.         {
  27.             // 如果该事件类型还没有订阅者，则创建新列表
  28.             if (!_subscribers.ContainsKey(eventType))
  29.             {
  30.                 _subscribers[eventType] = new List<object>();
  31.             }
  33.             // 添加处理委托到列表中 (确保不重复添加同一个委托实例)
  34.             if (!_subscribers[eventType].Contains(handler))
  35.             {
  36.                  _subscribers[eventType].Add(handler);
  37.             }
  38.         }
  39.     }
  41.     public void Unsubscribe<TEvent>(Action<TEvent> handler) where TEvent : class
  42.     {
  43.         Type eventType = typeof(TEvent);
  45.         lock (_lock)
  46.         {
  47.             // 检查是否存在该事件类型的订阅列表，以及列表中是否包含该处理委托
  48.             if (_subscribers.TryGetValue(eventType, out var handlers))
  49.             {
  50.                 handlers.Remove(handler);
  52.                 // 如果移除后列表为空，可以选择从字典中移除该事件类型（可选，优化内存）
  53.                 if (handlers.Count == 0)
  54.                 {
  55.                     _subscribers.Remove(eventType);
  56.                 }
  57.             }
  58.         }
  59.     }
  61.     public void Publish<TEvent>(TEvent @event) where TEvent : class
  62.     {
  63.         Type eventType = typeof(TEvent);
  64.         List<object> handlersToInvoke = null;
  66.         lock (_lock)
  67.         {
  68.             // 检查是否有该事件类型的订阅者
  69.             if (_subscribers.TryGetValue(eventType, out var handlers))
  70.             {
  71.                 // 复制列表以避免在迭代时修改集合（如果在处理程序中取消订阅）
  72.                 // 并且在锁外执行委托调用，减少锁的持有时间
  73.                 handlersToInvoke = handlers.ToList();
  74.             }
  75.         }
  77.         // 在锁外部调用处理程序，避免长时间持有锁
  78.         if (handlersToInvoke != null)
  79.         {
  80.             foreach (var handlerObj in handlersToInvoke)
  81.             {
  82.                 // 将 object 转换回具体的 Action<TEvent>
  83.                 if (handlerObj is Action<TEvent> handler)
  84.                 {
  85.                     try
  86.                     {
  87.                         // 调用处理委托
  88.                         // 实际应用中可能需要考虑：
  89.                         // 1. 异步执行 (handler( @event).ConfigureAwait(false)等)
  90.                         // 2. 错误处理 (try-catch 并记录日志)
  91.                         handler(@event);
  92.                     }
  93.                     catch (Exception ex)
  94.                     {
  95.                         // 处理或记录订阅者处理事件时发生的异常
  96.                         Console.WriteLine($"处理事件 {@event.GetType().Name} 时出错: {ex.Message}");
  97.                         // 根据策略决定是否继续通知其他订阅者
  98.                     }
  99.                 }
  100.             }
  101.         }
  102.     }
  103. }
复制代码
4. 示例：发布者和订阅者
  1. // === 发布者 ===
  2. public class OrderService
  3. {
  4.     private readonly IEventBus _eventBus;
  6.     public OrderService(IEventBus eventBus)
  7.     {
  8.         _eventBus = eventBus ?? throw new ArgumentNullException(nameof(eventBus));
  9.     }
  11.     public void CreateOrder(int orderId)
  12.     {
  13.         Console.WriteLine($"订单服务: 正在创建订单 {orderId}...");
  14.         // ... 执行创建订单的逻辑 ...
  15.         Console.WriteLine($"订单服务: 订单 {orderId} 创建成功.");
  17.         // 创建事件实例
  18.         var orderCreatedEvent = new OrderCreatedEvent(orderId);
  20.         // 发布事件到 EventBus
  21.         Console.WriteLine($"订单服务: 发布 {nameof(OrderCreatedEvent)} 事件.");
  22.         _eventBus.Publish(orderCreatedEvent);
  23.     }
  24. }
  26. public class AuthenticationService
  27. {
  28.      private readonly IEventBus _eventBus;
  30.     public AuthenticationService(IEventBus eventBus)
  31.     {
  32.         _eventBus = eventBus ?? throw new ArgumentNullException(nameof(eventBus));
  33.     }
  35.      public void Login(string username)
  36.     {
  37.         Console.WriteLine($"认证服务: 用户 {username} 尝试登录...");
  38.         // ... 认证逻辑 ...
  39.         Console.WriteLine($"认证服务: 用户 {username} 登录成功.");
  41.         var loggedInEvent = new UserLoggedInEvent(username);
  42.         Console.WriteLine($"认证服务: 发布 {nameof(UserLoggedInEvent)} 事件.");
  43.         _eventBus.Publish(loggedInEvent);
  44.     }
  45. }
  48. // === 订阅者 ===
  49. public class EmailService : IDisposable // 实现 IDisposable 以便在服务生命周期结束时取消订阅
  50. {
  51.     private readonly IEventBus _eventBus;
  52.     // 保存委托实例，以便取消订阅时使用
  53.     private readonly Action<OrderCreatedEvent> _orderCreatedHandler;
  55.     public EmailService(IEventBus eventBus)
  56.     {
  57.         _eventBus = eventBus ?? throw new ArgumentNullException(nameof(eventBus));
  58.         // 将处理方法赋值给字段
  59.         _orderCreatedHandler = HandleOrderCreated;
  60.         // 订阅事件
  61.         _eventBus.Subscribe(_orderCreatedHandler); // 使用保存的委托实例
  62.         Console.WriteLine("邮件服务: 已订阅 OrderCreatedEvent.");
  63.     }
  65.     private void HandleOrderCreated(OrderCreatedEvent @event)
  66.     {
  67.         Console.WriteLine($"邮件服务: 收到订单创建事件. 准备发送邮件通知 (订单ID: {@event.OrderId})...");
  68.         // 模拟发送邮件
  69.         Thread.Sleep(50); // 模拟耗时
  70.         Console.WriteLine($"邮件服务: 订单 {@event.OrderId} 的创建通知邮件已发送.");
  71.     }
  73.     public void Dispose()
  74.     {
  75.         // 在对象销毁时取消订阅，防止内存泄漏
  76.         _eventBus.Unsubscribe(_orderCreatedHandler); // 使用保存的委托实例取消订阅
  77.         Console.WriteLine("邮件服务: 已取消订阅 OrderCreatedEvent.");
  78.     }
  79. }
  81. public class InventoryService
  82. {
  83.     private readonly IEventBus _eventBus;
  85.     public InventoryService(IEventBus eventBus)
  86.     {
  87.         _eventBus = eventBus ?? throw new ArgumentNullException(nameof(eventBus));
  88.         // 可以直接使用 Lambda 表达式或方法组，但取消订阅会稍微麻烦些
  89.         // 如果需要取消订阅，最好像 EmailService 那样保存委托实例
  90.         _eventBus.Subscribe<OrderCreatedEvent>(HandleOrderCreated);
  91.          Console.WriteLine("库存服务: 已订阅 OrderCreatedEvent.");
  92.     }
  94.     private void HandleOrderCreated(OrderCreatedEvent @event)
  95.     {
  96.         Console.WriteLine($"库存服务: 收到订单创建事件. 正在为订单 {@event.OrderId} 预留库存...");
  97.         // 模拟库存操作
  98.         Thread.Sleep(100); // 模拟耗时
  99.         Console.WriteLine($"库存服务: 订单 {@event.OrderId} 的库存已预留.");
  100.     }
  102.      // 注意：如果用 Lambda 或方法组直接订阅且需要取消订阅，需要保存那个委托实例
  103.      // public void StopListening() { _eventBus.Unsubscribe<OrderCreatedEvent>(HandleOrderCreated); } // 这样可能无法正确取消匿名方法
  104. }
  106. public class AuditService
  107. {
  108.      private readonly IEventBus _eventBus;
  110.     public AuditService(IEventBus eventBus)
  111.     {
  112.         _eventBus = eventBus;
  113.         // 订阅多种事件
  114.         _eventBus.Subscribe<OrderCreatedEvent>(LogEvent);
  115.         _eventBus.Subscribe<UserLoggedInEvent>(LogEvent);
  116.          Console.WriteLine("审计服务: 已订阅 OrderCreatedEvent 和 UserLoggedInEvent.");
  117.     }
  119.     // 一个通用的日志记录方法，处理不同类型的事件
  120.     private void LogEvent<TEvent>(TEvent @event) where TEvent : class
  121.     {
  122.          Console.WriteLine($"审计服务: 记录到事件 - 类型: {@event.GetType().Name}, 内容: {@event.ToString()}");
  123.     }
  125.     // 可选：如果需要取消订阅，同样需要保存委托实例
  126. }
复制代码
5. 运行示例
  1. using System;
  3. public class Program
  4. {
  5.     public static void Main(string[] args)
  6.     {
  7.         // 1. 创建 EventBus 实例 (通常通过依赖注入容器管理，这里手动创建)
  8.         IEventBus eventBus = new EventBus();
  10.         // 2. 创建订阅者实例，并将 EventBus 注入
  11.         using (var emailService = new EmailService(eventBus)) // 使用 using 确保 Dispose 被调用以取消订阅
  12.         {
  13.             var inventoryService = new InventoryService(eventBus);
  14.             var auditService = new AuditService(eventBus);
  16.             Console.WriteLine("\n--- 初始化完成, 准备发布事件 ---\n");
  18.             // 3. 创建发布者实例
  19.             var orderService = new OrderService(eventBus);
  20.             var authService = new AuthenticationService(eventBus);
  22.             // 4. 发布者执行操作并发布事件
  23.             orderService.CreateOrder(101);
  25.             Console.WriteLine("\n----------------------------\n");
  27.             authService.Login("Alice");
  29.             Console.WriteLine("\n--- 事件处理完成 ---\n");
  31.             // 模拟 EmailService 停止工作（离开 using 块会自动调用 Dispose 取消订阅）
  32.         } // emailService.Dispose() 会在这里被调用
  34.         Console.WriteLine("\n--- EmailService 已停止并取消订阅 ---\n");
  36.         // 再次创建订单，EmailService 不应再收到通知
  37.         var orderService2 = new OrderService(eventBus); // 可以复用之前的，这里新建仅为演示
  38.         orderService2.CreateOrder(102);
  41.         Console.WriteLine("\n--- 程序结束 ---");
  42.         Console.ReadKey();
  43.     }
  44. }
复制代码
输出示例:
  1. 邮件服务: 已订阅 OrderCreatedEvent.
  2. 库存服务: 已订阅 OrderCreatedEvent.
  3. 审计服务: 已订阅 OrderCreatedEvent 和 UserLoggedInEvent.
  5. --- 初始化完成, 准备发布事件 ---
  7. 订单服务: 正在创建订单 101...
  8. 订单服务: 订单 101 创建成功.
  9. 订单服务: 发布 OrderCreatedEvent 事件.
  10. 邮件服务: 收到订单创建事件. 准备发送邮件通知 (订单ID: 101)...
  11. 库存服务: 收到订单创建事件. 正在为订单 101 预留库存...
  12. 审计服务: 记录到事件 - 类型: OrderCreatedEvent, 内容: 订单已创建: ID=101, 时间=...
  13. 邮件服务: 订单 101 的创建通知邮件已发送.
  14. 库存服务: 订单 101 的库存已预留.
  16. ----------------------------
  18. 认证服务: 用户 Alice 尝试登录...
  19. 认证服务: 用户 Alice 登录成功.
  20. 认证服务: 发布 UserLoggedInEvent 事件.
  21. 审计服务: 记录到事件 - 类型: UserLoggedInEvent, 内容: 用户已登录: 用户名=Alice, 时间=...
  23. --- 事件处理完成 ---
  25. 邮件服务: 已取消订阅 OrderCreatedEvent.
  27. --- EmailService 已停止并取消订阅 ---
  29. 订单服务: 正在创建订单 102...
  30. 订单服务: 订单 102 创建成功.
  31. 订单服务: 发布 OrderCreatedEvent 事件.
  32. 库存服务: 收到订单创建事件. 正在为订单 102 预留库存...
  33. 审计服务: 记录到事件 - 类型: OrderCreatedEvent, 内容: 订单已创建: ID=102, 时间=...
  34. 库存服务: 订单 102 的库存已预留.
  36. --- 程序结束 ---
复制代码
关键点和考虑:

  • 线程安全: 上述实现使用了简单的 lock 来保护 _subscribers 字典。对于高并发场景，可能需要更精细的锁策略或使用 ConcurrentDictionary（但仍需注意其内部列表的线程安全）。在 Publish 方法中，先复制订阅者列表再在锁外调用委托，是为了减少锁的持有时间，提高并发性能，并防止在处理程序中修改订阅列表导致迭代错误。
  • 错误处理: 订阅者的处理程序中可能抛出异常。Publish 方法应该能够处理这些异常（例如，记录日志），并决定是否继续通知其他订阅者。
  • 异步处理: 如果订阅者的处理逻辑是 I/O 密集型或耗时的，应该考虑使用异步委托 (Func) 和异步发布机制，以避免阻塞发布线程。
  • 取消订阅: 非常重要，尤其是在订阅者的生命周期比 EventBus 短的情况下（例如 UI 组件、临时服务）。否则会导致内存泄漏（EventBus 持有对已不再需要的订阅者的引用）。使用 IDisposable 是一种常见的管理取消订阅的方式。确保 Unsubscribe 时传入的是 Subscribe 时使用的同一个委托实例。
  • 弱引用: 对于某些场景（如 UI 框架），如果订阅者可能被垃圾回收而没有显式取消订阅，可以使用 WeakReference 来持有委托，但这会增加实现的复杂性。
  • 依赖注入: 在实际应用中，IEventBus 通常注册为单例（Singleton）服务，并通过依赖注入（DI）容器注入到需要它的发布者和订阅者中。
  • 事件继承: 当前设计是基于精确的事件类型匹配。如果需要支持订阅基类事件或接口事件来接收所有派生类/实现类的事件，Publish 方法的逻辑需要相应调整（遍历事件类型的继承链或接口）。
这个简单的 EventBus 实现提供了一个基础框架，你可以根据具体项目的需求进行扩展和优化。

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

使用道具 举报
相关推荐
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册
发帖
胆饬
6 天前

0

粉丝关注

10

主题发布

板块介绍填写区域，请于后台编辑
微信扫一扫