EventBusIneShop—解析微软微服务架构Demo（四）

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

今天说下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(Func handler)
where T : IntegrationEvent
where TH : IIntegrationEventHandler;
void Unsubscribe()
where TH : IIntegrationEventHandler
where T : IntegrationEvent;
void Publish(IntegrationEvent @event);
}

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

private readonly ConcurrentDictionary> _eventAndHandlerMapping;

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

public void Subscribe(Func handler)
where T : IntegrationEvent
where TH : IIntegrationEventHandler
{
var eventName = typeof(T).Name;
var cOntainsKey= _subsManager.HasSubscriptionsForEvent();
if (!containsKey)
{
if (!_persistentConnection.IsConnected)
{
_persistentConnection.TryConnect();
}
using (var channel = _persistentConnection.CreateModel())
{
channel.QueueBind(queue: _queueName,
exchange: BROKER_NAME,
routingKey: eventName);
}
}
_subsManager.AddSubscription(handler);
}

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

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

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

在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之间的依赖关系，我们引用大神的总结：

在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>();
var orderStartedHandler = app.ApplicationServices
.GetService>();
var eventBus = app.ApplicationServices.GetRequiredService();
eventBus.Subscribe

(() => app.ApplicationServices.GetRequiredService

());
eventBus.Subscribe
(() => app.ApplicationServices.GetRequiredService());
}

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

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

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

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

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

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

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

QQ群：376248054

邮箱：james@taogame.com

QQ:785418

微信：jamesying1

QQ群：376248054 通关：cnblogs

技术改变生活，技术改变人生！用技术来创造价值，拥有技术，不仅仅是开发，您将获得更多！如果您觉得我能帮到您，您可以通过扫描下面二维码来【捐助】我！