分布式事件总线

分布式事件总线系统允许发布和订阅跨应用/服务边界传输的事件. 你可以使用分布式事件总线在微服务或应用程序之间异步发送和接收消息.

提供程序

分布式事件总线系统提供了一个可以被任何提供程序实现的抽象. 有四种开箱即用的提供程序:

  • LocalDistributedEventBus 是默认实现,实现作为进程内工作的分布式事件总线. 是的!如果没有配置真正的分布式提供程序,默认实现的工作方式与本地事件总线一样.
  • RabbitMqDistributedEventBus 通过RabbitMQ实现分布式事件总线. 请参阅RabbitMQ集成文档了解如何配置它.
  • KafkaDistributedEventBus 通过Kafka实现分布式事件总线. 请参阅Kafka集成文档了解如何配置它.
  • RebusDistributedEventBus 通过Rebus实现分布式事件总线. 请参阅Rebus集成文档了解如何配置它.

使用本地事件总线作为默认具有一些重要的优点. 最重要的是:它允许你编写与分布式体系结构兼容的代码. 您现在可以编写一个整体应用程序,以后可以拆分成微服务. 最好通过分布式事件而不是本地事件在边界上下文之间(或在应用程序模块之间)进行通信.

例如,预构建的应用模块被设计成在分布式系统中作为服务工作,同时它们也可以在独立应用程序中作为模块工作,而不依赖于外部消息代理.

发布事件

以下介绍了两种发布分布式事件的方法.

IDistributedEventBus

可以注入 IDistributedEventBus 并且使用发布分布式事件.

示例: 产品的存货数量发生变化时发布分布式事件

  1. using System;
  2. using System.Threading.Tasks;
  3. using Volo.Abp.DependencyInjection;
  4. using Volo.Abp.EventBus.Distributed;
  6. namespace AbpDemo
  7. {
  8.     public class MyService : ITransientDependency
  9.     {
  10.         private readonly IDistributedEventBus _distributedEventBus;
  12.         public MyService(IDistributedEventBus distributedEventBus)
  13.         {
  14.             _distributedEventBus = distributedEventBus;
  15.         }
  17.         public virtual async Task ChangeStockCountAsync(Guid productId, int newCount)
  18.         {
  19.             await _distributedEventBus.PublishAsync(
  20.                 new StockCountChangedEto
  21.                 {
  22.                     ProductId = productId,
  23.                     NewCount = newCount
  24.                 }
  25.             );
  26.         }
  27.     }
  28. }

PublishAsync 方法需要一个参数:事件对象,它负责保持与事件相关的数据,是一个简单的普通类:

  1. using System;
  3. namespace AbpDemo
  4. {
  5.     [EventName("MyApp.Product.StockChange")]
  6.     public class StockCountChangedEto
  7.     {
  8.         public Guid ProductId { get; set; }
  10.         public int NewCount { get; set; }
  11.     }
  12. }

即使你不需要传输任何数据也需要创建一个类(在这种情况下为空类).

Eto 是我们按照约定使用的Event Transfer Objects(事件传输对象)的后缀. s虽然这不是必需的,但我们发现识别这样的事件类很有用(就像应用层上的DTO 一样).

事件名称

EventNameattribute是可选的,但建议使用. 如果不声明,事件名将事件名称将是事件类的全名. 这里是 AbpDemo.StockCountChangedEto.

关于序列化的事件对象

事件传输对象必须是可序列化的,因为将其传输到进程外时，它们将被序列化/反序列化为JSON或其他格式.

避免循环引用,多态,私有setter,并提供默认(空)构造函数,如果你有其他的构造函数.(虽然某些序列化器可能会正常工作),就像DTO一样.

实体/聚合根类

实体不能通过依赖注入注入服务,但是在实体/聚合根类中发布分布式事件是非常常见的.

示例: 在聚合根方法内发布分布式事件

  1. using System;
  2. using Volo.Abp.Domain.Entities;
  4. namespace AbpDemo
  5. {
  6.     public class Product : AggregateRoot<Guid>
  7.     {
  8.         public string Name { get; set; }
  10.         public int StockCount { get; private set; }
  12.         private Product() { }
  14.         public Product(Guid id, string name)
  15.             : base(id)
  16.         {
  17.             Name = name;
  18.         }
  20.         public void ChangeStockCount(int newCount)
  21.         {
  22.             StockCount = newCount;
  24.             //ADD an EVENT TO BE PUBLISHED
  25.             AddDistributedEvent(
  26.                 new StockCountChangedEto
  27.                 {
  28.                     ProductId = Id,
  29.                     NewCount = newCount
  30.                 }
  31.             );
  32.         }
  33.     }
  34. }

AggregateRoot 类定义了 AddDistributedEvent 来添加一个新的分布式事件,事件在聚合根对象保存(创建,更新或删除)到数据库时发布.

如果实体发布这样的事件,以可控的方式更改相关属性是一个好的实践,就像上面的示例一样 - StockCount只能由保证发布事件的 ChangeStockCount 方法来更改.

IGeneratesDomainEvents 接口

实际上添加分布式事件并不是 AggregateRoot 类独有的. 你可以为任何实体类实现 IGeneratesDomainEvents. 但是 AggregateRoot 默认实现了它简化你的工作.

不建议为不是聚合根的实体实现此接口,因为它可能不适用于此类实体的某些数据库提供程序. 例如它适用于EF Core,但不适用于MongoDB.

它是如何实现的?

调用 AddDistributedEvent 不会立即发布事件. 当你将更改保存到数据库时发布该事件;

  • 对于 EF Core, 它在 DbContext.SaveChanges 中发布.
  • 对于 MongoDB, 它在你调用仓储的 InsertAsync, UpdateAsync 或 DeleteAsync 方法时发由 (因为MongoDB没有更改跟踪系统).

订阅事件

一个服务可以实现 IDistributedEventHandler<TEvent> 来处理事件.

示例: 处理上面定义的StockCountChangedEto

  1. using System.Threading.Tasks;
  2. using Volo.Abp.DependencyInjection;
  3. using Volo.Abp.EventBus.Distributed;
  5. namespace AbpDemo
  6. {
  7.     public class MyHandler
  8.         : IDistributedEventHandler<StockCountChangedEto>,
  9.           ITransientDependency
  10.     {
  11.         public async Task HandleEventAsync(StockCountChangedEto eventData)
  12.         {
  13.             var productId = eventData.ProductId;
  14.         }
  15.     }
  16. }

这就是全部.

  • MyHandler 由ABP框架自动发现,并在发生 StockCountChangedEto 事件时调用 HandleEventAsync.
  • 如果你使用的是分布式消息代理,比如RabbitMQ,ABP会自动订阅消息代理上的事件,获取消息执行处理程序.
  • 如果事件处理程序成功执行(没有抛出任何异常),它将向消息代理发送确认(ACK).

你可以在处理程序注入任何服务来执行所需的逻辑. 一个事件处理程序可以订阅多个事件,但是需要为每个事件实现 IDistributedEventHandler<TEvent> 接口.

事件处理程序类必须注册到依赖注入(DI),示例中使用了 ITransientDependency. 参阅DI文档了解更多选项.

预定义的事件

如果你配置,ABP框架会为实体自动发布创建,更新和删除分布式事件.

事件类型

有三种预定义的事件类型:

  • EntityCreatedEto<T> 是实体 T 创建后发布.
  • EntityUpdatedEto<T> 是实体 T 更新后发布.
  • EntityDeletedEto<T> 是实体 T 删除后发布.

这些都是泛型的, T 实际上是Event Transfer Object (ETO)的类型,而不是实体的类型,因为实体对象不能做为事件数据传输,所以通常会为实体类定义一个ETO类,如为 Product 实体定义 ProductEto.

订阅事件

订阅自动事件与订阅常规分布式事件相同.

示例: 产品更新后获取通知

  1. using System.Threading.Tasks;
  2. using Volo.Abp.DependencyInjection;
  3. using Volo.Abp.Domain.Entities.Events.Distributed;
  4. using Volo.Abp.EventBus.Distributed;
  6. namespace AbpDemo
  7. {
  8.     public class MyHandler : 
  9.         IDistributedEventHandler<EntityUpdatedEto<ProductEto>>,
  10.         ITransientDependency
  11.     {
  12.         public async Task HandleEventAsync(EntityUpdatedEto<ProductEto> eventData)
  13.         {
  14.             var productId = eventData.Entity.Id;
  15.             //TODO
  16.         }
  17.     }
  18. }
  • MyHandler 实现了 IDistributedEventHandler<EntityUpdatedEto<ProductEto>>.

配置

你可以在模块的 ConfigureServices 中配置 AbpDistributedEntityEventOptions添加选择器.

示例: 配置示例

  1. Configure<AbpDistributedEntityEventOptions>(options =>
  2. {
  3.     //Enable for all entities
  4.     options.AutoEventSelectors.AddAll();
  6.     //Enable for a single entity
  7.     options.AutoEventSelectors.Add<IdentityUser>();
  9.     //Enable for all entities in a namespace (and child namespaces)
  10.     options.AutoEventSelectors.AddNamespace("Volo.Abp.Identity");
  12.     //Custom predicate expression that should return true to select a type
  13.     options.AutoEventSelectors.Add(
  14.         type => type.Namespace.StartsWith("MyProject.")
  15.     );
  16. });
  • 最后一个提供了灵活性来决定是否应该针对给定的实体类型发布事件. 返回 true 代表为该 Type 发布事件.

你可以添加多个选择器. 如果选择器之一与实体类型匹配,则将其选中.

事件传输对象

一旦你为一个实体启用了自动事件,ABP框架就会为实体上的更改发布事件. 如果你没有为实体指定对应的Event Transfer Object(ETO), ABP框架会使用一个标准类型 EntityEto,它只有两个属性:

  • EntityType (string): 实体类的全名(包括命令空间).
  • KeysAsString (string): 已更改实体的主键.如果它只有一个主键,这个属性将是主键值. 对于复合键,它包含所有用,(逗号)分隔的键.

因此可以实现 IDistributedEventHandler<EntityUpdatedEto<EntityEto>> 订阅事件. 但是订阅这样的通用事件不是一个好方法,你可以为实体类型定义对应的ETO.

示例: 为 Product 声明使用 ProductEto

  1. Configure<AbpDistributedEntityEventOptions>(options =>
  2. {
  3.     options.AutoEventSelectors.Add<Product>();
  4.     options.EtoMappings.Add<Product, ProductEto>();
  5. });

在这个示例中;

  • 添加选择器允许发布 Product 实体的创建,更新和删除事件.
  • 配置为使用 ProductEto 作为事件传输对象来发布与 Product 相关的事件.

分布式事件系统使用对象到对象的映射系统来映射 Product 对象到 ProductEto 对象,你需要配置映射. 请参阅可以对象到对象映射文档了解所有选项,下面的示例展示了如何使用AutoMapper库配置它.

示例: 使用AutoMapper配置 Product 到 ProductEto 映射

  1. using System;
  2. using AutoMapper;
  3. using Volo.Abp.Domain.Entities.Events.Distributed;
  5. namespace AbpDemo
  6. {
  7.     [AutoMap(typeof(Product))]
  8.     public class ProductEto : EntityEto
  9.     {
  10.         public Guid Id { get; set; }
  11.         public string Name { get; set; }
  12.     }
  13. }

此示例使用AutoMapper的 AutoMap 属性配置的映射. 你可以创建一个配置文件类代替. 请参阅AutoMapper文档了解更多选项.