EventBus In eShop -- 解析微软微服务架构Demo(四)

本文涉及的产品
注册配置 MSE Nacos/ZooKeeper,118元/月
服务治理 MSE Sentinel/OpenSergo,Agent数量 不受限
云原生网关 MSE Higress,422元/月
简介: 引言 大家好像对分析源码厌倦了,说实在我也会厌倦,不过不看是无法分析其后面的东西,从易到难是一个必要的过程。 今天说下EventBus,前几天园里的大神已经把其解刨,我今天就借着大神的肩膀,分析下在eShop项目中EventBus的实现。

引言

大家好像对分析源码厌倦了,说实在我也会厌倦,不过不看是无法分析其后面的东西,从易到难是一个必要的过程。

今天说下EventBus,前几天园里的大神已经把其解刨,我今天就借着大神的肩膀,分析下在eShop项目中EventBus的实现。

最近发觉转发文章不写出处的,特此加上链接:http://inday.cnblogs.com

解析源码

我们知道使用EventBus是为了解除Publisher和Subscriber之间的依赖性,这样我们的Publisher就不需要知道有多少Subscribers,只需要通过EventBus进行注册管理就好了,在eShop项目中,有一个这样的接口IEventBus(eShopOnContainers\src\BuildingBlocks\EventBus\EventBus\Abstractions)

public interface IEventBus
{
    void Subscribe<T, TH>(Func<TH> handler)
        where T : IntegrationEvent
        where TH : IIntegrationEventHandler<T>;
    void Unsubscribe<T, TH>()
        where TH : IIntegrationEventHandler<T>
        where T : IntegrationEvent;

    void Publish(IntegrationEvent @event);
}

我们可以看到这个接口定义了EventBus所需的一些操作, 对比大神的EventBus,相关功能都是一致的,我们看下它的实现类:EventBusRabbitMQ,从名字上可以看出,这是一个通过RabbitMQ来进行管理的EventBus,我们可以看到它使用了IEventBusSubscriptionsManager进行订阅存储,也就是大神文中的:

private readonly ConcurrentDictionary<Type, List<Type>> _eventAndHandlerMapping;

微软在Demo中把其提取出了接口,把一些常用方法给提炼了出来,但是核心还是Dictionary<string, List<Delegate>>, 使用Dictionary进行Map映射。通过Subscribe和UnSubscribe进行订阅和取消,使用Publish方法进行发布操作。

public void Subscribe<T, TH>(Func<TH> handler)
    where T : IntegrationEvent
    where TH : IIntegrationEventHandler<T>
{
    var eventName = typeof(T).Name;
    var containsKey = _subsManager.HasSubscriptionsForEvent<T>();
    if (!containsKey)
    {
        if (!_persistentConnection.IsConnected)
        {
            _persistentConnection.TryConnect();
        }

        using (var channel = _persistentConnection.CreateModel())
        {
            channel.QueueBind(queue: _queueName,
                                exchange: BROKER_NAME,
                                routingKey: eventName);
        }
    }

    _subsManager.AddSubscription<T, TH>(handler);

}

我们看到在订阅的时候,EventBus会检查下在Map中是否有相应的注册,如果没有的话首先回去RabbitMQ中创建一个新的channel进行绑定,随后在Map中进行注册映射。

UnSubscribe则直接从Map中取消映射,通过OnEventRemoved事件判断Map下此映射的subscriber是否为空,为空则从RabbitMQ中关闭channel。

在RabbitMQ的构造方法中,我们看到这样一个创建:CreateConsumerChannel(),这里创建了一个EventingBasicConsumer,当Queue中有新的消息时会通过ProcessEvent执行Map中注册的handler(subscribers),看图可能更清晰些:

 

Event In eShop

在ProcessEvent方法中,回去Map中找寻subscribers,然后通过动态反射进行执行:

private async Task ProcessEvent(string eventName, string message)
{

    if (_subsManager.HasSubscriptionsForEvent(eventName))
    { 
        var eventType = _subsManager.GetEventTypeByName(eventName);
        var integrationEvent = JsonConvert.DeserializeObject(message, eventType);
        var handlers = _subsManager.GetHandlersForEvent(eventName);

        foreach (var handlerfactory in handlers)
        {
            var handler = handlerfactory.DynamicInvoke();
            var concreteType = typeof(IIntegrationEventHandler<>).MakeGenericType(eventType);
            await (Task)concreteType.GetMethod("Handle").Invoke(handler, new object[] { integrationEvent });
        }
    }
}

微软通过简单的代码解耦了Publisher和Subscribers之间的依赖关系,我们引用大神的总结:

image

应用

在catalog.api中,微软出现了EventBus,我在上一篇中也提到了,这是我的一个疑惑,因为在catalog中并没有订阅操作,直接执行了Publish操作,原先以为是一个空操作,后来看了Basket.Api我才知道为何微软要用RabbitMQ。

使用RabbitMQ,我们不仅是从类之间的解耦,更可以跨项目,跨语言,跨平台的解耦,publisher仅仅需要把消息体(IntegrationEvent)传送到RabbitMQ,Consumer从Queue中获取消息体,然后推送到Subscribers执行相应的操作。我们看下Basket.Api.Startup.cs:

protected virtual void ConfigureEventBus(IApplicationBuilder app)
{
    var catalogPriceHandler = app.ApplicationServices
        .GetService<IIntegrationEventHandler<ProductPriceChangedIntegrationEvent>>();

    var orderStartedHandler = app.ApplicationServices
        .GetService<IIntegrationEventHandler<OrderStartedIntegrationEvent>>();

    var eventBus = app.ApplicationServices.GetRequiredService<IEventBus>();

    eventBus.Subscribe<ProductPriceChangedIntegrationEvent, ProductPriceChangedIntegrationEventHandler>
        (() => app.ApplicationServices.GetRequiredService<ProductPriceChangedIntegrationEventHandler>());

    eventBus.Subscribe<OrderStartedIntegrationEvent, OrderStartedIntegrationEventHandler>
        (() => app.ApplicationServices.GetRequiredService<OrderStartedIntegrationEventHandler>());
}

在这个方法里,我们看到了Subscribe操作,想想之前的提问有点搞笑,不过研究明白了也不错,对吧!

总结

今天我们看了EventBus在Demo中的应用,总结一下。

1、EventBus可以很好的解耦订阅者和发布者之间的依赖

2、使用RabbitMQ能够跨项目、跨平台、跨语言的解耦订阅者和发布者

虽然在Demo中我们看到对订阅者的管理是通过Dictionary内存的方式,所以我们的Subscribe仅仅只在Basket.Api中看到,但微软是通过IEventBusSubscriptionsManager接口定义的,我们可以通过自己的需求来进行定制,可以做成分布式的,比如使用memcached。

写在最后

每个月到下旬就会比较忙,所以文章发布会比较慢,但我也会坚持学习完eShop的,为了学习,我建了个群,大家可以进来一起学习,有什么建议和问题都可以进来哦。

eShop虽好,但不建议大家放到生产环境,毕竟是一个Demo,而且目前还是ALPHA版本,用来学习是一个很好的教材,这就是一个大杂烩,学习中你会学到很多新的东西,大家如果看好core的发展,可以一起研究下。

QQ群:376248054

相关文章
|
19天前
|
存储 JSON 数据库
Elasticsearch 分布式架构解析
【9月更文第2天】Elasticsearch 是一个分布式的搜索和分析引擎,以其高可扩展性和实时性著称。它基于 Lucene 开发,但提供了更高级别的抽象,使得开发者能够轻松地构建复杂的搜索应用。本文将深入探讨 Elasticsearch 的分布式存储和检索机制,解释其背后的原理及其优势。
69 5
|
6天前
|
设计模式 Java 关系型数据库
【Java笔记+踩坑汇总】Java基础+JavaWeb+SSM+SpringBoot+SpringCloud+瑞吉外卖/谷粒商城/学成在线+设计模式+面试题汇总+性能调优/架构设计+源码解析
本文是“Java学习路线”专栏的导航文章,目标是为Java初学者和初中高级工程师提供一套完整的Java学习路线。
|
2天前
|
负载均衡 5G 网络性能优化
深入解析LTE(长期演进技术)的基本架构及其关键组件
深入解析LTE(长期演进技术)的基本架构及其关键组件
17 2
|
5天前
|
算法 Linux 调度
操作系统的心脏:现代操作系统架构的深度解析
本文深入探讨了现代操作系统的架构设计,重点分析了进程管理、内存管理和文件系统等核心组件。通过对Linux和Windows两大主流操作系统的比较,揭示了不同设计哲学在实际应用中的表现。旨在为操作系统开发者提供参考,同时帮助普通用户更好地理解其设备背后的软件机制。 ##
18 2
|
20天前
|
设计模式 存储 人工智能
深度解析Unity游戏开发:从零构建可扩展与可维护的游戏架构,让你的游戏项目在模块化设计、脚本对象运用及状态模式处理中焕发新生,实现高效迭代与团队协作的完美平衡之路
【9月更文挑战第1天】游戏开发中的架构设计是项目成功的关键。良好的架构能提升开发效率并确保项目的长期可维护性和可扩展性。在使用Unity引擎时,合理的架构尤为重要。本文探讨了如何在Unity中实现可扩展且易维护的游戏架构,包括模块化设计、使用脚本对象管理数据、应用设计模式(如状态模式)及采用MVC/MVVM架构模式。通过这些方法,可以显著提高开发效率和游戏质量。例如,模块化设计将游戏拆分为独立模块。
44 3
|
2天前
|
搜索推荐 Linux Android开发
深入解析安卓与iOS系统架构设计差异
本文旨在探讨Android和iOS两大主流操作系统在架构设计上的根本差异。通过分析两种系统的设计理念、核心组件以及实际应用表现,揭示它们如何反映不同的开发哲学和用户体验策略。我们将从系统层级结构、内存管理机制、用户界面设计三个方面入手,逐一对比Android的开放性和灵活性如何与其对手iOS的封闭性和一致性相互辉映。
|
20天前
|
开发者 Java
Play Framework深度解析:依赖注入的神秘力量,如何助力Web应用架构优化?答案即将揭晓!
【8月更文挑战第31天】依赖注入(DI)是现代软件开发的关键技术,用于分离对象创建与依赖关系,提升代码的可维护性和可测试性。Play Framework是一款高性能Java Web框架,内置了基于Google Guice的DI支持。本文探讨Play Framework中DI的最佳实践,包括定义组件、构造函数注入、字段注入以及作用域控制和自定义绑定等高级特性,帮助开发者轻松构建结构清晰、可维护性高的Web应用。
29 0
|
22天前
|
监控 网络协议 Java
Tomcat源码解析】整体架构组成及核心组件
Tomcat,原名Catalina,是一款优雅轻盈的Web服务器,自4.x版本起扩展了JSP、EL等功能,超越了单纯的Servlet容器范畴。Servlet是Sun公司为Java编程Web应用制定的规范,Tomcat作为Servlet容器,负责构建Request与Response对象,并执行业务逻辑。
Tomcat源码解析】整体架构组成及核心组件
|
1月前
|
存储 NoSQL Redis
redis 6源码解析之 object
redis 6源码解析之 object
55 6
|
6天前
|
存储 缓存 Java
什么是线程池?从底层源码入手,深度解析线程池的工作原理
本文从底层源码入手,深度解析ThreadPoolExecutor底层源码,包括其核心字段、内部类和重要方法,另外对Executors工具类下的四种自带线程池源码进行解释。 阅读本文后,可以对线程池的工作原理、七大参数、生命周期、拒绝策略等内容拥有更深入的认识。
什么是线程池?从底层源码入手,深度解析线程池的工作原理

推荐镜像

更多