EventBus/EventQueue 再思考

来源:https://www.cnblogs.com/weihanli/archive/2020/05/23/12941919.html

EventBus/EventQueue 再思考 Intro 之前寫過兩篇文章,造輪子系列的 /`EventQueue EvenStore` 可能有點誤解,有興趣可以參考 /, 最近把 Event 相關的邏輯做了一個重構,修改 ,引入了 ,重新設計了 Event 相關的組件 重構後的 Event Ev ...


EventBus/EventQueue 再思考

Intro

之前寫過兩篇文章,造輪子系列的 EventBus/EventQueue,回想起來覺得當前的想法有點問題,當時對 EvenStore 可能有點誤解,有興趣可以參考 https://www.cnblogs.com/weihanli/p/implement-a-simple-event-bus.html/https://www.cnblogs.com/weihanli/p/implement-event-queue.html

最近把 Event 相關的邏輯做了一個重構,修改 EventStore,引入了 IEventHandlerFactory,重新設計了 Event 相關的組件

重構後的 Event

  • Event: 事件的抽象定義
  • EventHandler:事件處理器抽象定義
  • EventHandlerFactory:事件處理器工廠,用來根據事件類型獲取事件處理器(新增)
  • EventPublisher:事件發佈器,用於事件發佈
  • EventSubscriber:事件訂閱器,用於管理事件的訂閱
  • EventSubscriptionManager:事件訂閱管理器,在 EventSubscriber 的基礎上增加了一個根據事件類型獲取事件訂閱器類型的方法
  • EventBus:事件匯流排,由 EventPubliser 和 EventSubscriber 組合而成,用來比較方便的做事件發佈和訂閱
  • EventQueue:事件隊列,希望某些消息順序處理的時候可以考慮用 EventQueue 的模式
  • EventStore:事件存儲,事件的持久化存儲(在之前的版本里,EventStore 實際作用是一個 EventSubscriptionManager,在最近的版本更新中已修改)

以上 EventSubscriberEventSubscriptionManager 一般不直接用,一般用 EventBus 來處理即可

EventHandlerFactory

這次引入了 EventHandlerFactory 用來抽象獲取 EventHandler 的邏輯,原來的設計里是在處理 Event 的時候獲取 EventHandler 的類型,然後從依賴註入框架中獲取或創建新的 event handler 實例之後再調用 EventHandler 的 Handle 方法處理事件,有一些冗餘

使用 EventHandlerFactory 之後就可以直接獲取一個 EventHandler 實例集合,具體是實例化還是從依賴註入中獲取就由 EventHandlerFactory 來決定了,這樣就可以對依賴註入很友好,對於基於記憶體的簡單 EventBus 來說,在服務註冊之後就不需要再調用 Subscribe 去顯式訂閱了,因為再註冊服務的時候就已經隱式實現了訂閱的邏輯,這樣實際就不需要 EventSubscriptionManager 來管理訂閱了,訂閱信息都在依賴註入框架內部,比如說 CounterEvent,要獲取它的訂閱信息,我只需要從依賴註入框架中獲取 IEventHandler<CounterEvent> 的實例即可,實際就代替了原先 “EventStoreInMemory”,現在的 EventSubscriptionManagerInMemory

基於依賴註入的 EventHandlerFactory 定義:

public sealed class DependencyInjectionEventHandlerFactory : IEventHandlerFactory
{
    private readonly IServiceProvider _serviceProvider;

    public DependencyInjectionEventHandlerFactory(IServiceProvider serviceProvider = null)
    {
        _serviceProvider = serviceProvider ?? DependencyResolver.Current;
    }

    public ICollection<IEventHandler> GetHandlers(Type eventType)
    {
        var eventHandlerType = typeof(IEventHandler<>).MakeGenericType(eventType);
        return _serviceProvider.GetServices(eventHandlerType).Cast<IEventHandler>().ToArray();
    }
}

如果不使用依賴註入,也可以根據 IEventSubscriptionManager 訂閱信息來實現:

public sealed class DefaultEventHandlerFactory : IEventHandlerFactory
{
    private readonly IEventSubscriptionManager _subscriptionManager;
    private readonly ConcurrentDictionary<Type, ICollection<IEventHandler>> _eventHandlers = new ConcurrentDictionary<Type, ICollection<IEventHandler>>();
    private readonly IServiceProvider _serviceProvider;

    public DefaultEventHandlerFactory(IEventSubscriptionManager subscriptionManager, IServiceProvider serviceProvider = null)
    {
        _subscriptionManager = subscriptionManager;
        _serviceProvider = serviceProvider ?? DependencyResolver.Current;
    }

    public ICollection<IEventHandler> GetHandlers(Type eventType)
    {
        var eventHandlers = _eventHandlers.GetOrAdd(eventType, type =>
        {
            var handlerTypes = _subscriptionManager.GetEventHandlerTypes(type);
            var handlers = handlerTypes
                .Select(t => (IEventHandler)_serviceProvider.GetServiceOrCreateInstance(t))
                .ToArray();
            return handlers;
        });
        return eventHandlers;
    }
}

EventQueue Demo

來看一下 EventQueue 的示例,示例基於 asp.net core 的,定義了一個 HostedService 來實現一個 EventConsumer 來消費 EventQueue 中的事件信息

EventConsumer 定義如下:

public class EventConsumer : BackgroundService
{
    private readonly IEventQueue _eventQueue;
    private readonly IEventHandlerFactory _eventHandlerFactory;

    public EventConsumer(IEventQueue eventQueue, IEventHandlerFactory eventHandlerFactory)
    {
        _eventQueue = eventQueue;
        _eventHandlerFactory = eventHandlerFactory;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            var queues = await _eventQueue.GetQueuesAsync();
            if (queues.Count > 0)
            {
                await queues.Select(async q =>
                        {
                            var @event = await _eventQueue.DequeueAsync(q);
                            if (null != @event)
                            {
                                var handlers = _eventHandlerFactory.GetHandlers(@event.GetType());
                                if (handlers.Count > 0)
                                {
                                    await handlers
                                            .Select(h => h.Handle(@event))
                                            .WhenAll()
                                        ;
                                }
                            }
                        })
                        .WhenAll()
                    ;
            }

            await Task.Delay(1000, stoppingToken);
        }
    }
}

定義 PageViewEventPageViewEventHandler,用來記錄和處理請求訪問記錄

public class PageViewEvent : EventBase
{
}

public class PageViewEventHandler : EventHandlerBase<PageViewEvent>
{
    public static int Count;

    public override Task Handle(PageViewEvent @event)
    {
        Interlocked.Increment(ref Count);
        return Task.CompletedTask;
    }
}

事件很簡單,事件處理也只是增加了 PageViewEventHandler 內定義的 Count。

服務註冊:

// 註冊事件核心組件
// 會註冊 EventBus、EventHandlerFactory、EventQueue 等
services.AddEvents()
    // 註冊 EventHanlder 
    .AddEventHandler<PageViewEvent, PageViewEventHandler>()
    ;
// 註冊 EventQueuePubliser,預設註冊的 IEventPublisher 是 EventBus
services.AddSingleton<IEventPublisher, EventQueuePublisher>();
// 註冊 EventConsumer
services.AddHostedService<EventConsumer>();

事件發佈,定義了一個中間件來發佈 PageViewEvent,定義如下:

// pageView middleware
app.Use((context, next) =>
        {
            var eventPublisher = context.RequestServices.GetRequiredService<IEventPublisher>();
            eventPublisher.Publish(new PageViewEvent());

            return next();
        });

然後定義一個介面來獲取上面定義的 PageViewEventHandler 中的 Count

[Route("api/[controller]")]
public class EventsController : ControllerBase
{
    [HttpGet("pageViewCount")]
    public IActionResult Count()
    {
        return Ok(new { Count = PageViewEventHandler.Count });
    }
}

運行起來之後,訪問幾次介面,看上面的介面返回 Count 是否會增加,正常的話每訪問一次介面就會增加 1,併發訪問問題也不大,因為每個事件都是順序處理的,即使併發訪問也沒有關係,事件發佈之後,在隊列里都是順序處理的,這也就是引入事件隊列的目的(好像上面的原子遞增沒什麼用了...) 如果沒看到了增加,稍等一會兒再訪問試試,事件處理會遲到,但總會處理,畢竟是非同步處理的,有些延遲很正常,而且上面我們還有一個 1s 的延遲

More

更多關於上述 Event 相關的信息可以參考代碼: https://github.com/WeihanLi/WeihanLi.Common/tree/dev/src/WeihanLi.Common/Event

作者水平有限,如果上述有哪些不對的地方還望指出,萬分感謝

Reference


您的分享是我們最大的動力!

更多相關文章
  • 概念內部狀態、外部狀態、享元池角色 & UMLDemo: 編輯器圖片重用 - JavaReference概念享元模式(Flyweight Pattern),是以 共用 的方式,對 大量細粒度對象 重用,來減少記憶體的使用(避免大量重覆地創建、銷毀對象)。名稱中的Flyweight,是搏擊比賽中體重級別... ...
  • 聲明:這是在windows10上進行kafka_2.13demo搭建時的過程記錄,提供給同學們參考。 1.jdk先要裝一下。 2.先安裝zookeeper,這裡不贅述,貼一個鏈接 https://blog.csdn.net/ring300/article/details/80446918。記得測試一 ...
  • 前言 本文的文字及圖片來源於網路,僅供學習、交流使用,不具有任何商業用途,版權歸原作者所有,如有問題請及時聯繫我們以作處理。 用Python也差不多一年多了,Python應用最多的場景還是Web快速開發、爬蟲、自動化運維:寫過簡單網站、寫過自動發帖腳本、寫過收發郵件腳本、寫過簡單驗證碼識彆腳本。 爬 ...
  • 1、緩存穿透 原因: 指定查詢一個一定不存在的數據,我們去查詢某個商品,但是緩存中沒有,那麼就是直接查詢資料庫,高併發下假如同時100萬請求同時查詢,那麼他就會直接穿過緩存去查詢資料庫,那麼它將會導致資料庫崩潰無法工作【一直查詢一個不存在的結果,導致緩存一直不命中,全部來查詢資料庫。導致資料庫壓力過 ...
  • 最近是微軟開了Build 2020大會,由於疫情原因,改成了線上舉行,Build大會上,C 公佈9.0版本。 我個人對於C 的更新向來都是喜聞樂見,樂於接受的,對於博客園上某些人天天嘲諷C 只會增加語法糖的人,我向來對他們不屑一顧,認為他們是井底之蛙。 因此我仔細看了微軟發的文章 "Welcome ...
  • /// <summary /// C List轉換成DataTable /// </summary /// <param name="list" </param /// <returns </returns public static System.Data.DataTable ListToData ...
  • 0.前言 在上一篇中,我們知道瞭如何使用SqlSugar,但是也只是簡單的瞭解瞭如何使用,仿佛是套著鐐銬行走,這明顯不符合一個合格的程式員應有的素養。所以,這一篇我們將對其進行深挖,探究其背後的秘密。 1. 花式映射 在實際開發中,程式中的實體類和資料庫的表名並不能完全一致,造成的原因有很多,例如說 ...
  • 以前需要反編譯的時候,一直都是用的神器 Reflector。但畢竟還是有點不方便,因為要從 VS 切換到另一個工具。而且,Reflector 現在收費了。 然而現在才發現,原來 VS 已經自帶反編譯功能了,不愧是宇宙第一 IDE。 工具 → 選項 → 文本編輯器 → c 高級 → 啟用導航到反編譯源 ...
一周排行
  • 一:背景 1. 講故事 曾今在項目中發現有同事自定義結構體的時候,居然沒有重寫Equals方法,比如下麵這段代碼: static void Main(string[] args) { var list = Enumerable.Range(0, 1000).Select(m => new Point ...
  • 最近一個朋友有個關於素數的小東西要寫一下,素數是什麼呢?除了1和他本身不能被其他數整除,那麼這個數就是素數,1除外哦。我們知道概念那就很簡單了,直接代碼擼起。 ...
  • 前言 在開發編程中,我們經常會遇到功能非常相似的功能模塊,只是他們的處理的數據不一樣,所以我們會分別採用多個方法來處理不同的數據類型。但是這個時候,我們就會想一個問題,有沒有辦法實現利用同一個方法來傳遞不同種類型的參數呢? 這個時候,泛型也就因運而生,專門來解決這個問題的。 泛型是在C 2.0就推出 ...
  • 本文章主要用於介紹在Asp.Net Mvc(C#)中使用Fleck製作一個Html5的即時聊天室,含有完整代碼和演示Demo。 ...
  • 出庫單的功能。能學習了出庫單管理之後,WMS的 主體功能算是完成了。當然一個成熟的WMS還包括了盤點,報表,策略規則,移庫功能及與其他系統(ERP、TMS等)的介面,實現無縫集成,打破信息孤島,讓數據實時、準確和同步。 ...
  • Data StructureThere're two types of variables in C#, reference type and value type.Enum:enum Color{Red=0,Green=1}//equals to enum Color{Red,//start fr... ...
  • 0. 前言 該項目使用Maven進行管理和構建,所以需要預先配置好Maven。嗯,在這個系列里就不做過多的介紹了。 1. 創建項目 先創建一個pom.xml 文件,添加以下內容: <?xml version="1.0" encoding="UTF-8"?> <project xmlns="http: ...
  • API 概述 API(Application Programming Interface),應用程式編程介面。 Java API是一本程式員的 字典 ,是JDK中提供給我們使用的類的說明文檔。 這些類將底層的代碼實現封裝了起來,我們不需要關心這些類是如何實現的,只需要學習這些類如何使用即可。 所以我 ...
  • 女程式員是這麼徵婚的: SELECT * FROM 男人們 WHERE 未婚=true and 同性戀=false and 有房=true and 有車=true and 條件 in (帥氣,紳士,大度,氣質,智慧,溫柔,體貼,會浪漫,活潑,可愛,最好還能帶孩子) and 年齡 between(24 ...
  • 有很多剛學習軟體測試的小伙伴,都會在網路上找尋各種學習資料,去提升自己的專業技能水平。因此,我決定定期分享我整理收集的一些軟體測試的測試工具下載、面試寶典、視頻教學合集。都整理好了,有需要的可以關註我(獲取方式在文末) 軟體測試的學習,不止是基礎理論,還需要學習測試工具的用法,如介面工具Postma ...