内容简介:从上一篇文章大家可以看出,实现一个自己的消息总线框架是非常重要的内容,消息总线可以将界限上下文之间进行解耦,也可以为大并发访问提供必要的支持。1.界限上下文解耦:在DDD第一波文章中,当更新了订单信息后,我们通过调用经销商界限上下文的领域模型和仓储,进行了经销商信息的更新,这造成了耦合。通过一个消息总线,可以在订单界限上下文的WebApi服务(来源微服务-生产者)更新了订单信息后,发布一个事件消息到消息总线的某个队列中,经销商界限上下文的WebApi服务(消费者)订阅这个事件消息,然后交给自己的Handl
从上一篇文章大家可以看出,实现一个自己的消息总线框架是非常重要的内容,消息总线可以将界限上下文之间进行解耦,也可以为大并发访问提供必要的支持。
消息总线的作用:
1.界限上下文解耦:在DDD第一波文章中,当更新了订单信息后,我们通过调用经销商界限上下文的领域模型和仓储,进行了经销商信息的更新,这造成了耦合。通过一个消息总线,可以在订单界限上下文的WebApi服务(来源微服务-生产者)更新了订单信息后,发布一个事件消息到消息总线的某个队列中,经销商界限上下文的WebApi服务(消费者)订阅这个事件消息,然后交给自己的Handler进行消息处理,更新自己的经销商信息。这样就实现了订单界限上下文与经销商界限上下文解耦。
2.大并发支持:可以通过消息总线进一步提升下单的性能。我们可以将用户下单的操作直接交给一个下单命令WebApi接收,下单命令WebApi接收到命令后,直接丢给一个消息总线的队列,然后立即给前端返回下单结果。这样用户就不用等待后续的复杂订单业务逻辑,加快速度。后续订单的一系列处理交给消息的Handler进行后续的处理与消息的进一步投递。
消息总线设计重点:
1.定义消息(事件)的接口:所有需要投递与处理的消息,都从这个消息接口继承,因为需要约束消息中必须包含的内容,比如消息的ID、消息产生的时间等。
public interface IEvent { Guid Id { get; set; } DateTime CreateDate { get; set; } } 复制代码
2.定义消息(事件)处理器接口:当消息投递到消息总线队列中后,一定有消费者WebApi接收并处理这个消息,具体的处理方法逻辑在订阅方处理器中实现,这里先需要定义处理器的接口,便于在消息总线框架中使用。
public interface IEventHandler { Task<bool> HandleAsync<TEvent>(TEvent @event) where TEvent : IEvent; } 复制代码
从上面代码可以看出,消息(事件)处理器处理的类型就是从IEvent接口继承的消息类。
3.定义消息(事件)与消息(事件)处理器关联接口:一种类型的消息被投递后,一定要在订阅方找到这种消息的处理器进行处理,所以一定要定义二者的关联接口,这样才能将消息与消息处理器对应起来,才能实现消息被订阅后的处理。
public interface IEventHandlerExecutionContext { void RegisterEventHandler<TEvent, TEventHandler>() where TEvent : IEvent where TEventHandler : IEventHandler; bool IsRegisterEventHandler<TEvent, TEventHandler>() where TEvent : IEvent where TEventHandler : IEventHandler; Task HandleAsync<TEvent>(TEvent @event) where TEvent : IEvent; } 复制代码
RegisterEventHandler方法就是建立消息与消息处理器的关联,这个方法其实是在订阅方使用,订阅方告诉消息总线,什么样的消息应该交给我的哪个处理器进行处理。
IsRegisterEventHandler方法是判断消息与处理器之间是否已经存在关联。
HandleAsync方法是通过查找到消息对应的处理器后,然后调用处理器自己的Handle方法进行消息的处理。
4.定义消息发布、订阅与消息总线接口:消息总线至少要支持两个功能,一个是生产者能够发布消息到我的消息总线,另一个是订阅方需要能够从我这个消息总线订阅消息。
public interface IEventPublisher { void Publish<TEvent>(TEvent @event) where TEvent : IEvent; } 复制代码
从上面代码可以看出,生产者发布的消息仍然要从IEvent继承的类型。
public interface IEventSubscriber { void Subscribe<TEvent, TEventHandler>() where TEvent : IEvent where TEventHandler : IEventHandler; } 复制代码
上面代码是订阅方用于从消息总线订阅消息,从代码中可以看出,它的最终的实现其实就是建立消息与处理器之间的关联
public interface IEventBus:IEventPublisher,IEventSubscriber { } 复制代码
消息(事件)总线从两个接口继承下来,同时支持消息的发布与消息的订阅。
5.实现事件基类:上面已经订阅了消息(事件)的接口,这里来实现事件的基类,其实就是实现消息ID与产生的时间:
public class BaseEvent : IEvent { public Guid Id { get; set; } public DateTime CreateDate { get; set; } public BaseEvent() { this.Id = Guid.NewGuid(); this.CreateDate = DateTime.Now; } } 复制代码
6.实现消息总线基类:消息总线底层的依赖可以是各种消息代理产品,比如RabbitMq、Kafaka或第三方云平台提供的消息代理产品,通常我们要封装这些消息代理产品。在封装之前,我们需要定义顶层的消息总线基类实现,主要的目的是未来依赖于它的具体实现可替换,另外也将消息与消息处理器的关联接口传递进来,便于订阅方使用。
public abstract class BaseEventBus : IEventBus { protected readonly IEventHandlerExecutionContext eventHandlerExecutionContext; protected BaseEventBus(IEventHandlerExecutionContext eventHandlerExecutionContext) { this.eventHandlerExecutionContext = eventHandlerExecutionContext; } public abstract void Publish<TEvent>(TEvent @event) where TEvent : IEvent; public abstract void Subscribe<TEvent, TEventHandler>() where TEvent : IEvent where TEventHandler : IEventHandler; } 复制代码
7.实现消息与处理器关联:消息必须与处理器关联,订阅方收到特定类型的消息后,才知道交给哪个处理器处理。
class EventHandlerExecutionContext : IEventHandlerExecutionContext { private readonly IServiceCollection registry; private readonly IServiceProvider serviceprovider; private Dictionary<Type, List<Type>> registrations = new Dictionary<Type, List<Type>>(); public EventHandlerExecutionContext(IServiceCollection registry,Func<IServiceCollection, IServiceProvider> serviceProviderFactory = null) { this.registry = registry; this.serviceprovider = this.registry.BuildServiceProvider(); } //查找消息关联的处理器,然后调用处理器的处理方法 public async Task HandleAsync<TEvent>(TEvent @event) where TEvent : IEvent { var eventtype = @event.GetType(); if(registrations.TryGetValue(eventtype,out List<Type> handlertypes) && handlertypes.Count > 0) { using(var childscope = this.serviceprovider.CreateScope()) { foreach(var handlertype in handlertypes) { var handler = Activator.CreateInstance(handlertype) as IEventHandler; await handler.HandleAsync(@event); } } } } //判断消息与处理器之间是否有关联 public bool IsRegisterEventHandler<TEvent, TEventHandler>() where TEvent : IEvent where TEventHandler : IEventHandler 复制代码
//将消息与处理器关联起来,可以在内存中建立关联,也可以建立在数据库单独表中 public void RegisterEventHandler<TEvent, TEventHandler>() where TEvent : IEvent where TEventHandler : IEventHandler { Utils.DictionaryRegister(typeof(TEvent), typeof(TEventHandler), registrations); } } 复制代码
上面我们基本上就将消息总线的架子搭建起来了,也实现了基本的功能,下一章我们基于它来实现RabbitMq的消息总线。
QQ讨论群:309287205
微服务实战视频请关注微信公众号:MSSHCJ
以上就是本文的全部内容,希望本文的内容对大家的学习或者工作能带来一定的帮助,也希望大家多多支持 码农网
猜你喜欢:本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们。
Ajax Design Patterns
Michael Mahemoff / O'Reilly Media / 2006-06-29 / USD 44.99
Ajax, or Asynchronous JavaScript and XML, exploded onto the scene in the spring of 2005 and remains the hottest story among web developers. With its rich combination of technologies, Ajax provides a s......一起来看看 《Ajax Design Patterns》 这本书的介绍吧!