看eShopOnContainers学一个EventBus

简介: 最近在看微软eShopOnContainers 项目,看到事件总线觉得不错,和大家分享一下看完此文你将获得什么?eShop中是如何设计事件总线的实现一个InMemory事件总线eShop中是没有InMemory实现的,这算是一个小小小的挑战发布订阅模式发布订阅模式可以让应用程序组件之间解耦,这是我们使用这种模式最重要的理由之一,如果你完全不知道这个东西,建议你先通过搜索引擎了解一下这种模式,网上的资料很多这里就不再赘述了。

最近在看微软eShopOnContainers 项目,看到事件总线觉得不错,和大家分享一下

看完此文你将获得什么?

  1. eShop中是如何设计事件总线的
  2. 实现一个InMemory事件总线eShop中是没有InMemory实现的,这算是一个小小小的挑战

发布订阅模式

发布订阅模式可以让应用程序组件之间解耦,这是我们使用这种模式最重要的理由之一,如果你完全不知道这个东西,建议你先通过搜索引擎了解一下这种模式,网上的资料很多这里就不再赘述了。

eShop中的EventBus就是基于这种模式的发布/订阅
发布订阅模式核心概念有三个:发布者、订阅者、调度中心,这些概念在消息队列中就是生产者、消费者、MQ实例

在eShop中有两个EventBus的实现:

  • 基于RabbitMq的EventBusRabbitMQ
  • 基于AzureServiceBus的EventBusServiceBus

IEventBus开始

先来看一看,所有EventBus的接口IEventBus

public interface IEventBus
{
    void Publish(IntegrationEvent @event);

    void Subscribe<T, TH>()
        where T : IntegrationEvent
        where TH : IIntegrationEventHandler<T>;

    void SubscribeDynamic<TH>(string eventName)
        where TH : IDynamicIntegrationEventHandler;

    void UnsubscribeDynamic<TH>(string eventName)
        where TH : IDynamicIntegrationEventHandler;

    void Unsubscribe<T, TH>()
        where TH : IIntegrationEventHandler<T>
        where T : IntegrationEvent;
}

嗯,乍一看看是有点眼晕的,仔细看它的核心功能只有三个:

  1. Publish 发布
  2. Subscribe 订阅
  3. Unsubscribe 取消订阅

这对应着发布订阅模式的基本概念,不过对于事件总线的接口添加了许多约束:

  1. 发布的内容(消息)必须是IntegrationEvent及其子类
  2. 订阅事件必须指明要订阅事件的类型,并附带处理器类型
  3. 处理器必须是IIntegrationEventHandler的实现类

Ok,看到这里先不要管Dynamic相关的方法,然后记住这个两个关键点:

  1. 事件必须继承IntegrationEvent
  2. 处理器必须实现IIntegrationEventHandler<T>TIntegrationEvent子类

另外,看下 IntegrationEvent有什么

public class IntegrationEvent
{
    public IntegrationEvent()
    {
        Id = Guid.NewGuid();
        CreationDate = DateTime.UtcNow;
    }

    public Guid Id  { get; }
    public DateTime CreationDate { get; }
}

IEventBusSubscriptionsManager是什么

public interface IEventBusSubscriptionsManager
{
    bool IsEmpty { get; }
    event EventHandler<string> OnEventRemoved;
    void AddDynamicSubscription<TH>(string eventName)
       where TH : IDynamicIntegrationEventHandler;

    void AddSubscription<T, TH>()
       where T : IntegrationEvent
       where TH : IIntegrationEventHandler<T>;

    void RemoveSubscription<T, TH>()
         where TH : IIntegrationEventHandler<T>
         where T : IntegrationEvent;
    void RemoveDynamicSubscription<TH>(string eventName)
        where TH : IDynamicIntegrationEventHandler;

    bool HasSubscriptionsForEvent<T>() where T : IntegrationEvent;
    bool HasSubscriptionsForEvent(string eventName);
    Type GetEventTypeByName(string eventName);
    void Clear();
    IEnumerable<SubscriptionInfo> GetHandlersForEvent<T>() where T : IntegrationEvent;
    IEnumerable<SubscriptionInfo> GetHandlersForEvent(string eventName);
    string GetEventKey<T>();
}

这个接口看起来稍显复杂些,我们来简化下看看:

public interface IEventBusSubscriptionsManager
{
    void AddSubscription<T, TH>()
    void RemoveSubscription<T, TH>()
    IEnumerable<SubscriptionInfo> GetHandlersForEvent<T>() 
}

最终,这三个方法就是我们要关注的,添加订阅、移除订阅、获取指定事件的订阅信息。

SubscriptionInfo是什么?

public bool IsDynamic { get; }
public Type HandlerType{ get; }

SubscriptionInfo中只有两个信息,这是不是一个Dynamic类型的Event以及这个Event所对应的处理器的类型。

这是你可能会有另一个疑问:

这个和IEventBus有什么关系?

  1. IEventBusSubscriptionsManager含有更多功能:查看是否有订阅,获取事件的Type,获取事件的处理器等等
  2. IEventBusSubscriptionsManagerIEventBus使用,在RabbitMq和ServiceBus的实现中,都使用Manager去存储事件的信息,例如下面的代码:

     public void Subscribe<T, TH>()
         where T : IntegrationEvent
         where TH : IIntegrationEventHandler<T>
     {
         // 查询事件的全名
         var eventName = _subsManager.GetEventKey<T>();
    
         //向mq添加注册
         DoInternalSubscription(eventName);
    
         // 向manager添加订阅
         _subsManager.AddSubscription<T, TH>();
     }
    
     private void DoInternalSubscription(string eventName)
     {
         var containsKey = _subsManager.HasSubscriptionsForEvent(eventName);
         if (!containsKey)
         {
             if (!_persistentConnection.IsConnected)
             {
                 _persistentConnection.TryConnect();
             }
    
             using (var channel = _persistentConnection.CreateModel())
             {
                 channel.QueueBind(queue: _queueName,
                                     exchange: BROKER_NAME,
                                     routingKey: eventName);
             }
         }
     }

    查询事件的名字是manager做的,订阅的时候是先向mq添加订阅,之后又加到manager中,manager管理着订阅的基本信息。

另外一个重要功能是获取事件的处理器信息,在rabbit mq的实现中,ProcessEvent方法中用manager获取了事件的处理器,再用依赖注入获得处理器的实例,反射调用Handle方法处理事件信息:

    private async Task ProcessEvent(string eventName, string message)
    {
        // 从manager查询信息
        if (_subsManager.HasSubscriptionsForEvent(eventName))
        {
            using (var scope = _autofac.BeginLifetimeScope(AUTOFAC_SCOPE_NAME))
            {

                // 从manager获取处理器
                var subscriptions = _subsManager.GetHandlersForEvent(eventName);
                foreach (var subscription in subscriptions)
                {

                    // Di + 反射调用,处理事件(两个都是,只是针对是否是dynamic做了不同的处理)
                    if (subscription.IsDynamic)
                    { 
                        var handler = scope.ResolveOptional(subscription.HandlerType) as IDynamicIntegrationEventHandler;
                        dynamic eventData = JObject.Parse(message);
                        await handler.Handle(eventData);
                    }
                    else
                    {
                        var eventType = _subsManager.GetEventTypeByName(eventName);
                        var integrationEvent = JsonConvert.DeserializeObject(message, eventType);
                        var handler = scope.ResolveOptional(subscription.HandlerType);
                        var concreteType = typeof(IIntegrationEventHandler<>).MakeGenericType(eventType);
                        await (Task)concreteType.GetMethod("Handle").Invoke(handler, new object[] { integrationEvent });
                    }
                }
            }
        }
    }

IEventBusSubscriptionsManager的默认实现

在eShop中只有一个实现就是InMemoryEventBusSubscriptionsManager

这个类中有两个重要的字段

    private readonly Dictionary<string, List<SubscriptionInfo>> _handlers;
    private readonly List<Type> _eventTypes;

他们分别存储了事件列表和事件处理器信息词典

接下来就是实现一个

基于内存的事件总线

我们要做什么呢?IEventBusSubscriptionsManager 已经有了InMemory的实现了,我们可以直接拿来用,所以我们只需要自己实现一个EventBus就好了

先贴出最终代码:

public class InMemoryEventBus : IEventBus
{
    private readonly IServiceProvider _provider;
    private readonly ILogger<InMemoryEventBus> _logger;
    private readonly ISubscriptionsManager _manager;
    private readonly IList<IntegrationEvent> _events;
    public InMemoryEventBus(
        IServiceProvider provider,
        ILogger<InMemoryEventBus> logger, 
        ISubscriptionsManager manager)
    {
        _provider = provider;
        _logger = logger;
        _manager = manager;
    }

    public void Publish(IntegrationEvent e)
    {

        var eventType = e.GetType();
        var handlers = _manager.GetHandlersForEvent(eventType.FullName);

        foreach (var handlerInfo in handlers)
        {
            var handler = _provider.GetService(handlerInfo.HandlerType);

            var method = handlerInfo.HandlerType.GetMethod("Handle");

            method.Invoke(handler, new object[] { e });
        }
    }

    public void Subscribe<T, TH>()
        where T : IntegrationEvent
        where TH : IIntegrationEventHandler<T>
    {

        _manager.AddSubscription<T, TH>();

    }

    public void SubscribeDynamic<TH>(string eventName) where TH : IDynamicIntegrationEventHandler
    {
        throw new NotImplementedException();
    }

    public void Unsubscribe<T, TH>()
        where T : IntegrationEvent
        where TH : IIntegrationEventHandler<T>
    {
        _manager.RemoveSubscription<T, TH>();
    }

    public void UnsubscribeDynamic<TH>(string eventName) where TH : IDynamicIntegrationEventHandler
    {
        throw new NotImplementedException();
    }
}

首先构造函数中声明我们要使用的东西:

public InMemoryEventBus(
    IServiceProvider provider,
    ILogger<InMemoryEventBus> logger, 
    ISubscriptionsManager manager)
{
    _provider = provider;
    _logger = logger;
    _manager = manager;
}

这里要注意的就是IServiceProvider provider这是 DI容器,当我们在切实处理事件的时候我们选择从DI获取处理器的实例,而不是反射创建,这要做的好处在于,处理器可以依赖于其它东西,并且可以是单例的

public void Subscribe<T, TH>()
    where T : IntegrationEvent
    where TH : IIntegrationEventHandler<T>
{

    _manager.AddSubscription<T, TH>();

}

public void Unsubscribe<T, TH>()
    where T : IntegrationEvent
    where TH : IIntegrationEventHandler<T>
{
    _manager.RemoveSubscription<T, TH>();
}

订阅和取消订阅很简单,因为我们是InMemory的所以只调用了manager的方法。

接下来就是最重要的Publish方法,实现Publish有两种方式:

  1. 使用额外的线程和Queue让发布和处理异步
  2. 为了简单起见,我们先写个简单易懂的同步的

     public void Publish(IntegrationEvent e)
     {
         // 首先要拿到集成事件的Type信息
         var eventType = e.GetType();
    
         // 获取属于这个事件的处理器列表,可能有很多,注意获得的是SubscriptionInfo
         var handlers = _manager.GetHandlersForEvent(eventType.FullName);
    
         // 不解释循环
         foreach (var handlerInfo in handlers)
         {
             // 从DI中获取类型的实例
             var handler = _provider.GetService(handlerInfo.HandlerType);
    
             // 拿到Handle方法
             var method = handlerInfo.HandlerType.GetMethod("Handle");
    
             // 调用方法
             method.Invoke(handler, new object[] { e });
         }
     }

OK,我们的InMemoryEventBus就写好了!

要实践这个InMemoryEventBus,那么还需要一个IntegrationEvent的子类,和一个IIntegrationEventHandler<T>的实现类,这些都不难,例如我们做一个添加用户的事件,A在添加用户后,发起一个事件并将新用户的名字作为事件数据,B去订阅事件,并在自己的处理器中处理名字信息。

思路是这样的:

  1. 写一个 AddUserEvent:IntegrationEvent,里面有一个UserId和一个UserName
  2. 写一个AddUserEventHandler:IIntegrationEventHandler<AddUserEvent>,在Handle方法中输出UserId和Name到日志。
  3. 注册DI,你要注册下面这些服务:

     IEventBus=>InMemoryEventBus
     ISubscriptionsManager=>InMemorySubscriptionsManager
     AddUserEventHandler=>AddUserEventHandler
  4. 在Startup中为刚刚写的事件和处理器添加订阅(在这里已经可以获取到IEventBus实例了)
  5. 写一个Api接口或是什么,调用IEventBus的Publish方法,new 一个新的AddUserEvent作为参数传进去。

OK!到这里一个切实可用的InMemoryEventBus就可以使用了。

相关实践学习
5分钟轻松打造应对流量洪峰的稳定商城交易系统
本实验通过SAE极速部署一个微服务电商商城,同时结合RocketMQ异步解耦、削峰填谷的能力,带大家体验面对流量洪峰仍旧稳定可靠的商城交易系统!
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
目录
相关文章
|
10月前
|
人工智能 自然语言处理 安全
Claude怎么使用?claude中文官网才是最佳选择!
Claude AI 是由 Anthropic 公司潜心研发的划时代 AI 助手,致力于打造更安全、更友好、更可靠的 AI 系统,引领智能交互进入全新时代。
|
6月前
|
消息中间件 存储 Cloud Native
云消息队列 Kafka 版 V3 系列荣获信通院“云原生技术创新标杆案例”
2024 年 12 月 24 日,由中国信息通信研究院(以下简称“中国信通院”)主办的“2025 中国信通院深度观察报告会:算力互联网分论坛”,在北京隆重召开。本次论坛以“算力互联网 新质生产力”为主题,全面展示中国信通院在算力互联网产业领域的研究、实践与业界共识,与产业先行者共同探索算力互联网产业未来发展的方向。会议公布了“2024 年度云原生与应用现代化标杆案例”评选结果,“云消息队列 Kafka 版 V3 系列”荣获“云原生技术创新标杆案例”。
175 52
|
监控 Java 数据库连接
SpringBoot四大核心组件,必知必会!
SpringBoot四大核心组件,必知必会!
SpringBoot四大核心组件,必知必会!
|
3月前
|
存储 人工智能 安全
阿里云双项入选首批智算一体化权威评估 以AI Stack加速政企智能化升级 ——万卡智算集群服务推进方阵(ICCPA)第三期沙龙在京举办
2024年4月9日,中国信通院主办的智算集群服务沙龙第三期在京召开。阿里云凭借领先的AI技术能力,成为首批通过《面向大模型的智算一体化解决方案》评估的云厂商,并入选行业应用案例。会上,阿里云AI Stack赋能政企大模型高效落地,提供软硬一体推理优化框架,支持主流开源模型快速适配,助力企业构建高性能私有化AI服务,已在政务、金融等领域广泛应用。
199 6
|
11月前
|
存储 Ubuntu Linux
如何安装和使用 Docker:入门指南
如何安装和使用 Docker:入门指南
313 1
|
11月前
|
机器学习/深度学习 人工智能 自然语言处理
探索AI在文本生成中的应用
【8月更文挑战第1天】本文将深入探讨人工智能(AI)在文本生成领域的应用,包括技术原理、实现方法以及实际应用案例。我们将通过代码示例,展示如何使用机器学习和自然语言处理技术来生成高质量的文本内容。无论你是AI初学者还是专业人士,这篇文章都将为你提供有价值的信息和启发。
|
Java 数据库连接 数据库
Flink Connector JDBC已经被移到了一个独立的仓库
【2月更文挑战第23天】Flink Connector JDBC已经被移到了一个独立的仓库
177 1
|
JSON Ubuntu Linux
Windows 11 安装 Docker Desktop
WSL 全称是,适用于 Linux 的 Windows 子系统,可让开发人员按原样运行GNU/Linux 环境,包括大多数命令行工具、实用工具和应用程序,且不会产生传统虚拟机或双启动设置开销。简单的说就是WSL能让你在不安装 Linux 或者 VM(虚拟机)的情况下体验双系统!关于 WSL 更多信息,请查看
3209 2
Windows 11 安装 Docker Desktop
|
安全 关系型数据库 MySQL
MySQL里面的innodb_support_xa
在看innodb_support_xa之前我们要先看下什么是XA,什么是两阶段提交。 XA 一个协调分布式事务的标准接口,为了遵从ACID原则,允许多个DB参与事务。为了查看更多,请 Section 13.3.7, “XA Transactions”。
5893 0
|
机器学习/深度学习 Kubernetes 算法
浅析自动机器学习(AutoML)工具NNI(上)
NNI 简介 NNI (Neural Network Intelligence) 是一个轻量级但功能强大的自动机器学习(AutoML)工具包,可帮助用户自动化特征工程、神经架构搜索、超参数调优和模型压缩,并支持单机、本地多机、云等不同的运行环境。