看eShopOnContainers学一个EventBus

最近在看微软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>且T是IntegrationEvent子类

另外，看下 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. IEventBusSubscriptionsManager由IEventBus使用，在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");

        // 调用方法
        await (Task) 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就可以使用了。

原文地址：https://www.cnblogs.com/rocketRobin/p/8510198.html

时间： 2024-11-05 18:48:30

看eShopOnContainers学一个EventBus的相关文章

开机动画 看图学Android---Android 开发实例教程五

Android实例图解教程目录 http://blog.csdn.net/wyx100/article/details/45061407 在<开发一个新的android界面.界面跳转 看图学Android---Android 开发实例教程三.四>基础完成开机动画,开机界面的图片,在5.5秒内逐渐消失,显示主界面. 只修改StartActivity.java中代码以下部分 StartActivity.java源代码 package com.example.helloword; import an

每天进步一点点-实例为导学-一个java对象序列化的例子

序列化和反序列化例子 如果我们想要序列化一个对象, (对象 转 序列)首先要创建某些OutputStream(如FileOutputStream.ByteArrayOutputStream等),然后将这些OutputStream封装在一个ObjectOutputStream中.这时候,只需要调用writeObject()方法就可以将对象序列化,并将其发送给OutputStream(记住:对象的序列化是基于字节(1字节8位)的,不能使用Reader和Writer等基于字符的层次结构).而反序列的过

看漫画学 Redux (转)

1.看漫画学 Redux:http://www.oschina.net/p/a-cartoon-intro-to-redux-cn 2.Redux 核心概念 :http://www.jianshu.com/p/3334467e4b32 3.

http://www.driveword.com 驱动单词 看图学英语 看图学单词 一图一词

驱动单词正式发布,http://www.driveword.com 同时支持Mobile, Laptop, Desktop Computer 驱动单词,看图学单词,看图学英语,一图一词,单词一对一 截图欣赏

字母汉子组合的验证码，包括实现看不清换一个的功能

1.写一个验证码,前台什么都不用写,这是CreateVerificationCode_Page.aspx.cs后台代码 using System;using System.Collections;using System.Configuration;using System.Data;using System.Linq;using System.Web;using System.Web.Security;using System.Web.UI;using System.Web.UI.HtmlCo

开发一个新的android界面、界面跳转 看图学Android---Android 开发实例教程三、四

Android实例图解教程目录 http://blog.csdn.net/wyx100/article/details/45061407 一.课程功能 本课程讲述建立一个新界面和界面切换(从界面一切换到界面二). 二.课程界面 界面一(启动界面) 界面二(主界面) 三.工作流程 完成页面切换需要2个过程: 1.建立一个工程,见第二节. http://blog.csdn.net/wyx100/article/details/45248209 可以在该项目基础继续开发. 2.建立开机界面 先引入资源

一步一步图示开发第一个Android项目并运行 看图学Android---Android 开发实例教程二

一.生成一个Android应用工程 1.运行eclipse.exe 2.见下图 3.进入界面 4.输入 应用.项目.包名称 5.选择 项目建立的工作空间路径: 6.配置 图标和背景等特性 7.选择 Activity  的页面格式 8.设置对应名称 9.生成项目 二. 项目介绍 1.界面 可以使用工具采用拖拉的方式 拖动控件.  图形界面开发环境. 2.界面布局的代码格式 3.MainActivity.java  主程序结构   代码和视图绑定 3.代码MainActivity.java pack

Android消息传递之基于RxJava实现一个EventBus - RxBus（四）

前言: 上篇文章学习了Android事件总线管理开源框架EventBus,EventBus的出现大大降低了开发成本以及开发难度,今天我们就利用目前大红大紫的RxJava来实现一下类似EventBus事件总线管理,现在很多人都在说用这种方式来替代EventBus,今天我们从开发效率,开发难度等维度来分析一下到底能不能取代EventBus? 先回顾一下什么是EventBus?请看这篇文章Android消息传递之EventBus 3.0使用详解(三) 需求: 虽然软件工程师就害怕听见“需求”这两个字,

看武侠学编程_以九宫格为例介绍强大的声明式语言Prolog

如果要给众多编程语言分个类,你可能会把它们分成低级语言和高级语言,或者分成面向对象语言和面向过程语言.然而,更多中国程序员所不太熟悉的另外一种划分方式将会把计算机语言分成命令式和声明式两大阵营.之所以说大家可能不太熟悉这种划分,那是因为我们平常所使用绝大部分语言都是命令式的.但事实上你确实也应该注意到另外一大阵营的存在. 命令式编程(ImperativeProgramming)是现今最为广泛使用的编程范型.读者所熟知的众多计算机语言,如C.C++.Java.Pascal.Basic.Python