C# 如何实现一个事件总线

news/2024/5/19 23:06:41 标签: c#, .netcore, .net

EventBus（事件总线）是一种用于在应用程序内部或跨应用程序组件之间进行事件通信的机制。

它允许不同的组件通过发布和订阅事件来进行解耦和通信。在给定的代码片段中，我们可以看到一个使用C#实现的Event Bus。它定义了一些接口和类来实现事件的发布和订阅。

首先，我们有两个基本的约束接口：IEvent和IAsyncEventHandler<TEvent>。

IEvent是一个空接口，用于约束事件的类型。IAsyncEventHandler<TEvent>是一个泛型接口，用于约束事件处理程序的类型。它定义了处理事件的异步方法HandleAsync和处理异常的方法HandleException。接下来，我们有一个IEventBus接口，它定义了一些操作方法用于发布和订阅事件。

其中，Publish<TEvent>和PublishAsync<TEvent>方法用于发布事件，而OnSubscribe<TEvent>方法用于订阅事件。然后，我们看到一个实现了本地事件总线的类LocalEventBusManager<TEvent>。它实现了ILocalEventBusManager<TEvent>接口，用于在单一管道内处理本地事件。它使用了一个Channel<TEvent>来存储事件，并提供了发布事件的方法Publish和PublishAsync。此外，它还提供了一个自动处理事件的方法AutoHandle。

总的来说Event Bus提供了一种方便的方式来实现组件之间的松耦合通信。

通过发布和订阅事件，组件可以独立地进行操作，而不需要直接依赖于彼此的实现细节。

这种机制可以提高代码的可维护性和可扩展性。

Github仓库地址：https://github.com/DonPangPang/soda-event-bus

实现一些基本约束

先实现一些约束，实现IEvent约束事件，实现IAsyncEvnetHandler<TEvent> where TEvent:IEvent来约束事件的处理程序。

public interface IEvent
{

}

public interface IAsyncEventHandler<in TEvent> where TEvent : IEvent
{
    Task HandleAsync(IEvent @event);

    void HandleException(IEvent @event, Exception ex);
}

接下来规定一下咱们的IEventBus，会有哪些操作方法。基本就是发布和订阅。

public interface IEventBus
{
    void Publish<TEvent>(TEvent @event) where TEvent : IEvent;
    Task PublishAsync<TEvent>(TEvent @event) where TEvent : IEvent;

    void OnSubscribe<TEvent>() where TEvent : IEvent;
}

实现一个本地事件总线

本地事件处理

本地事件的处理我打算采用两种方式实现，一种是LocalEventBusManager即本地事件管理，第二种是LocalEventBusPool池化本地事件。

LocalEvnetBusManager

LocalEventBusManager主要在单一管道内进行处理，集中进行消费。

public interface ILocalEventBusManager<in TEvent>where TEvent : IEvent
{
    void Publish(TEvent @event);
    Task PublishAsync(TEvent @event) ;
    
    void AutoHandle();
}

public class LocalEventBusManager<TEvent>(IServiceProvider serviceProvider):ILocalEventBusManager<TEvent>
    where TEvent: IEvent
{
    readonly IServiceProvider _servicesProvider = serviceProvider;

    private readonly Channel<TEvent> _eventChannel = Channel.CreateUnbounded<TEvent>();

    public void Publish(TEvent @event)
    {
        Debug.Assert(_eventChannel != null, nameof(_eventChannel) + " != null");
        _eventChannel.Writer.WriteAsync(@event);
    }

    private CancellationTokenSource Cts { get; } = new();

    public void Cancel()
    {
        Cts.Cancel();
    }
    
    public async Task PublishAsync(TEvent @event)
    {
        await _eventChannel.Writer.WriteAsync(@event);
    }

    public void AutoHandle()
    {
        // 确保只启动一次
        if (!Cts.IsCancellationRequested) return;

        Task.Run(async () =>
        {
            while (!Cts.IsCancellationRequested)
            {
                var reader = await _eventChannel.Reader.ReadAsync();
                await HandleAsync(reader);
            }
        }, Cts.Token);
    }

    async Task HandleAsync(TEvent @event)
    {
        var handler = _servicesProvider.GetService<IAsyncEventHandler<TEvent>>();

        if (handler is null)
        {
            throw new NullReferenceException($"No handler for event {@event.GetType().Name}");
        }
        try
        {
            await handler.HandleAsync(@event);
        }
        catch (Exception ex)
        {
            handler.HandleException( @event, ex);
        }
    }
}
LocalEventBusPool

LocalEventBusPool即所有的Event都会有一个单独的管道处理，单独消费处理，并行能力更好一些。

public sealed class LocalEventBusPool(IServiceProvider serviceProvider)
{
    private readonly IServiceProvider _serviceProvider = serviceProvider;

    private class ChannelKey
    {
        public required string Key { get; init; }
        public int Subscribers { get; set; }

        public override bool Equals(object? obj)
        {
            if (obj is ChannelKey key)
            {
                return string.Equals(key.Key, Key, StringComparison.OrdinalIgnoreCase);
            }

            return false;
        }

        public override int GetHashCode()
        {
            return 0;
        }
    }

    private Channel<IEvent> Rent(string channel)
    {
        _channels.TryGetValue(new ChannelKey() { Key = channel }, out var value);

        if (value != null) return value;
        value = Channel.CreateUnbounded<IEvent>();
        _channels.TryAdd(new ChannelKey() { Key = channel }, value);
        return value;
    }

    private Channel<IEvent> Rent(ChannelKey channelKey)
    {
        _channels.TryGetValue(channelKey, out var value);
        if (value != null) return value;
        value = Channel.CreateUnbounded<IEvent>();
        _channels.TryAdd(channelKey, value);
        return value;
    }

    private readonly ConcurrentDictionary<ChannelKey, Channel<IEvent>> _channels = new();

    private CancellationTokenSource Cts { get; } = new();

    public void Cancel()
    {
        Cts.Cancel();
        _channels.Clear();
        Cts.TryReset();
    }

    public async Task PublishAsync<TEvent>(TEvent @event) where TEvent : IEvent
    {
        await Rent(typeof(TEvent).Name).Writer.WriteAsync(@event);
    }

    public void Publish<TEvent>(TEvent @event) where TEvent : IEvent
    {
        Rent(typeof(TEvent).Name).Writer.TryWrite(@event);
    }

    public void OnSubscribe<TEvent>() where TEvent : IEvent
    {
        var channelKey = _channels.FirstOrDefault(x => x.Key.Key == typeof(TEvent).Name).Key ??
                         new ChannelKey() { Key = typeof(TEvent).Name };
        channelKey.Subscribers++;

        Task.Run(async () =>
        {
            try
            {
                while (!Cts.IsCancellationRequested)
                {
                    var @event = await ReadAsync(channelKey);

                    var handler = _serviceProvider.GetService<IAsyncEventHandler<TEvent>>();
                    if (handler == null) throw new NullReferenceException($"No handler for Event {typeof(TEvent).Name}");
                    try
                    {
                        await handler.HandleAsync((TEvent)@event);
                    }
                    catch (Exception ex)
                    {
                        handler.HandleException((TEvent)@event, ex);
                    }
                }
            }
            catch (Exception e)
            {
                throw new InvalidOperationException("Error on onSubscribe handler", e);
            }
        }, Cts.Token);
    }

    private async Task<IEvent> ReadAsync(string channel)
    {
        return await Rent(channel).Reader.ReadAsync(Cts.Token);
    }

    private async Task<IEvent> ReadAsync(ChannelKey channel)
    {
        return await Rent(channel).Reader.ReadAsync(Cts.Token);
    }
}
LocalEventBus

实现LocalEventBus继承自IEventBus即可，如果有需要扩展的方法自行添加，池化和管理器的情况单独处理。

public interface ILocalEventBus: IEventBus
{

}
public class LocalEventBus(IServiceProvider serviceProvider, LocalEventBusOptions options) : ILocalEventBus
{
    private  LocalEventBusPool? EventBusPool => serviceProvider.GetService<LocalEventBusPool>();
    
    
    public void Publish<TEvent>(TEvent @event) where TEvent : IEvent
    {
        if (options.Pool)
        {
            Debug.Assert(EventBusPool != null, nameof(EventBusPool) + " != null");
            EventBusPool.Publish(@event);
        }
        else
        {
            var manager = serviceProvider.GetService<LocalEventBusManager<TEvent>>();
            if (manager is null) throw new NullReferenceException($"No manager for event {typeof(TEvent).Name}, please add singleton service it.");
            manager.Publish(@event);
        }
    }

    public async Task PublishAsync<TEvent>(TEvent @event) where TEvent : IEvent
    {
        if (options.Pool)
        {
            Debug.Assert(EventBusPool != null, nameof(EventBusPool) + " != null");
            await EventBusPool.PublishAsync(@event);
        }
        else
        {
            var manager = serviceProvider.GetService<LocalEventBusManager<TEvent>>();
            if (manager is null) throw new NullReferenceException($"No manager for event {typeof(TEvent).Name}, please add singleton service it.");
            await manager.PublishAsync(@event);
        }
    }

    public void OnSubscribe<TEvent>() where TEvent : IEvent
    {
        if (options.Pool)
        {
            Debug.Assert(EventBusPool != null, nameof(EventBusPool) + " != null");
            EventBusPool.OnSubscribe<TEvent>();
        }
        else
        {
            var manager = serviceProvider.GetService<LocalEventBusManager<TEvent>>();
            if (manager is null) throw new NullReferenceException($"No manager for event {typeof(TEvent).Name}, please add singleton service it.");
            manager.AutoHandle();
        }
    }
}

分布式事件总线

根据需要扩展即可，基本逻辑相同，但可能需要增加确认机制等。


http://www.niftyadmin.cn/n/5380409.html

相关文章

不知道不怪你：优秀网页UI的八大标准

不知道不怪你：优秀网页UI的八大标准

hello&#xff0c;我是大千UI工场&#xff0c;本期分享如何设计漂亮的网页UI&#xff0c;先从优秀网页UI的标准讲起&#xff0c; 欢迎评论、转发&#xff0c;有需求可以私信接单。 优秀网页UI设计的八大标准如下&#xff1a; 一致性 网页UI设计应保持一致性&#xff0c;包括颜…
阅读更多...
淘宝/天猫获得淘宝商品类目 API 的应用与优势

淘宝/天猫获得淘宝商品类目 API 的应用与优势

一、引言 在电子商务领域&#xff0c;淘宝和天猫无疑是中国最具代表性的电商平台。对于商家而言&#xff0c;如何有效地管理商品、提高销售效率&#xff0c;成为了日常运营中不可或缺的一部分。淘宝/天猫提供的商品类目 API 为商家提供了极大的便利&#xff0c;使得商家可以更…
阅读更多...
Vue的一些基础设置

Vue的一些基础设置

1.浏览器控制台显示Vue 设置找到扩展&#xff0c;搜索Vue 下载这个 然后 点击扩展按钮 点击详细信息 选择这个&#xff0c;然后重启一下就好了 ——————————————————————————————————————————— 2.优化工程结构 src的components里要…
阅读更多...
【C++】C++11上

【C++】C++11上

C11上 1.C11简介2.统一的列表初始化2.1 {} 初始化2.2 initializer_list 3.变量类型推导3.1auto3.2decltype3.3nullptr 4.范围for循环5.final与override6.智能指针7. STL中一些变化8.右值引用和移动语义8.1左值引用和右值引用8.2左值引用与右值引用比较8.3右值引用使用场景和意义…
阅读更多...
Windows环境部署nginx 文件服务器

Windows环境部署nginx 文件服务器

文章目录 前言一、pandas是什么&#xff1f;二、使用步骤 1.引入库2.读入数据总结 前言 在Windows环境下使用nginx部署简单的文件服务器 一、版本 1. Windows 使用版本 2. nginx 使用版本 选择Mainline Version版本 二、nginx配置 1. 下载 https://nginx.org/en/download.…
阅读更多...
一篇文章带你彻底搞懂Unicode、UTF-8、GB2312、GBK之间的关系

一篇文章带你彻底搞懂Unicode、UTF-8、GB2312、GBK之间的关系

前言 这是一篇深入浅出的技术文章&#xff0c;详细介绍了GBK、GB2312和UTF-8三种常见字符编码的特点及彼此之间的关系。通过阅读本文&#xff0c;你将了解到这些编码的背景、原理以及在实际应用中的差异和优势。无论你是初学者还是有一定编码基础的开发者&#xff0c;本文都能…
阅读更多...
Leecode之面试题消失的数字

Leecode之面试题消失的数字

一.题目及剖析 https://leetcode.cn/problems/missing-number-lcci/description/ 数组nums包含从0到n的所有整数&#xff0c;但其中缺了一个。请编写代码找出那个缺失的整数。你有办法在O(n)时间内完成吗&#xff1f; 注意&#xff1a;本题相对书上原题稍作改动 示例 1&…
阅读更多...
Linux常见指令（二）

Linux常见指令（二）

目录 常见指令 1.1more指令 1.2less指令 1.3tail指令 1.4时间相关的指令 1.5Cal指令 1.6find指令 1.7grep指令 1.8zip/unzip指令 1.9tar指令&#xff08;重要&#xff09; 1.10bc指令 常见指令 1.1more指令 语法 &#xff1a; more [ 选项 ][ 文件 ] 功能 &#xf…
阅读更多...
最新文章

Copyright @ 2022~2023