实现一个事件总线

这篇具有很好参考价值的文章主要介绍了实现一个事件总线。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

使用 C# 实现一个 Event Bus

Event Bus(事件总线)是一种用于在应用程序内部或跨应用程序组件之间进行事件通信的机制。它允许不同的组件通过发布和订阅事件来进行解耦和通信。

在给定的代码片段中,我们可以看到一个使用C#实现的Event Bus。它定义了一些接口和类来实现事件的发布和订阅。

首先,我们有两个基本的约束接口:IEventIAsyncEventHandler<TEvent>。IEvent是一个空接口,用于约束事件的类型。IAsyncEventHandler<TEvent>是一个泛型接口,用于约束事件处理程序的类型。它定义了处理事件的异步方法HandleAsync和处理异常的方法HandleException。

接下来,我们有一个IEventBus接口,它定义了一些操作方法用于发布和订阅事件。其中,Publish<TEvent>PublishAsync<TEvent>方法用于发布事件,而OnSubscribe<TEvent>方法用于订阅事件。

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

总的来说,Event Bus提供了一种方便的方式来实现组件之间的松耦合通信。通过发布和订阅事件,组件可以独立地进行操作,而不需要直接依赖于彼此的实现细节。这种机制可以提高代码的可维护性和可扩展性。

Github仓库地址: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();
        }
    }
}

分布式事件总线

根据需要扩展即可,基本逻辑相同,但可能需要增加确认机制等。文章来源地址https://www.toymoban.com/news/detail-771483.html

到了这里,关于实现一个事件总线的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • flutter开发实战-事件总线EventBus实现

    flutter开发实战-事件总线EventBus实现 在开发中,经常会需要一个广播机制,用以跨Widget事件通知。 事件总线 实现了订阅者模式,订阅者模式包含发布者和订阅者两种角色,可以通过事件总线来触发事件和监听事件。 实现eventBus 在工程的pubspec.yaml引入库 1.使用event_bus库 创建一

    2024年02月15日
    浏览(43)
  • Vue面试题:如何使用事件总线进行组件间数据传输?

    问题: 假设你正在为一个大型企业级Vue应用程序编写代码,并且需要在多个组件之间传递数据。其中一个组件需要从另一个组件中获取数据,但是这个组件可能还没有加载完成。在这种情况下,你会采取什么措施来确保数据的可靠传输?请提供代码示例来解释你的解决方案。

    2023年04月09日
    浏览(40)
  • 如何在 Android 应用中使用 RecyclerView 实现一个列表显示,并实现点击事件?

    首先,需要在项目的 build.gradle 文件中添加 RecyclerView 的依赖: 接下来,在布局文件中添加 RecyclerView: 接着,需要创建一个 Adapter 类,用于将数据绑定到 RecyclerView 上,如下所示: 在 onBindViewHolder() 方法中,我们可以将数据绑定到 ViewHolder 中的视图上。 需要注意的是,在 V

    2024年02月05日
    浏览(52)
  • c#事件(event)

    C#中的事件是一种特殊的委托,它用于实现观察者模式,允许对象在特定事件发生时通知其他对象。 以下是使用C#事件的示例: 首先,定义一个包含事件的类: 在上面的代码中,我们定义了一个EventPublisher类,其中包含一个名为MyEvent的事件。该事件基于EventHandler委托类型,它

    2024年02月10日
    浏览(45)
  • C#事件event

    事件模型的5个组成部分 事件拥有者(event source)(类对象)(有些书将其称为事件发布者) 事件成员(event)(事件拥有者的成员)(事件成员就是事件本身,事件不会主动发生,其只会在事件拥有者的内部逻辑的触发下发生。) 事件响应者(event subscriber)(类对象)(有

    2024年02月09日
    浏览(51)
  • 使用Vue.js框架的指令和事件绑定实现一个购物车的页面布局

    使用了v-model指令来实现全选/全不选的功能,当全选框被点击时,isAllChecked的值会被改变。 使用了v-if指令来判断购物车中是否有商品,如果有商品则渲染商品列表,否则显示购物车为空的提示。 使用了v-for指令来遍历datalist数组,渲染每个商品项。 使用了@change事件来监听商

    2024年02月12日
    浏览(55)
  • Tkinter Event事件处理方法

    在Tkinter中,事件处理方法是指在用户与GUI交互时,程序响应用户的操作并执行相应的操作。以下是一些常用的事件处理方法: Button-Click事件处理方法:当用户单击按钮时,执行相应的操作。 Key-Press事件处理方法:当用户按下键盘上的某个键时,执行相应的操作。 Mouse-Click事

    2024年02月09日
    浏览(46)
  • C#事件(event)的理解

    遇到一个开发的问题? 面试者:以面向对象的思想实现一下的场景: 猫:Miao一声,紧接着引发了一系列的行为~ Miao:引发了一系列的动作; 从代码层面来说:代码这样写好吗? 猫职责不单一(猫就是猫,他的行为只有Miao一声) 依赖太重,依赖了很多的普通类; 被依赖的类

    2024年03月09日
    浏览(60)
  • 【solidity三】Solidity中继承、创建合约和外部调用、合约事件event、solidity中元祖的使用、new、delete关键字的使用、solisity创建合约和外部调用

    - 最重要的两个全局变量 msg.sender 每一次和以太坊交互时都会产生一笔交易,这笔交易的执行人就是msg.sender。简而言之:谁调用的,msg.sender就是谁,每笔交易的msg.sender都可以不同。举例: 部署合约的时候,msg.sender就是部署的账户。 调用setMessage时,msg.sender就是调用账户。

    2024年02月03日
    浏览(46)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包