乐于分享
好东西不私藏

ABP 事件总线源码解读:ILocalEventBus 的订阅-发布机制

ABP 事件总线源码解读:ILocalEventBus 的订阅-发布机制

源码位置

framework/src/Volo.Abp.EventBus.Abstractions/Volo/Abp/EventBus/
framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Local/LocalEventBus.cs

事件总线的架构

ABP 的事件总线分两层:

IEventBus(基接口)
  ├── ILocalEventBus(本地,进程内)
  └── IDistributedEventBus(分布式,跨进程)

本文专注于 ILocalEventBus。

订阅机制

方式一:自动订阅(推荐)

public class OrderCreatedHandler : ILocalEventHandler<OrderCreatedEvent>, ITransientDependency
{
public async Task HandleEventAsync(OrderCreatedEvent eventData)
 { }
}

ABP 启动时,EventHandlerAutoRegistrar 扫描所有实现了 ILocalEventHandler<> 且注册了 ITransientDependency 的类型,自动调用 localEventBus.Subscribe<TEvent, THandler>()。

方式二:手动订阅

localEventBus.Subscribe<OrderCreatedEvent>(async eventData =>
{
    // 处理逻辑

});

Subscribe 返回 IDisposable,调用 Dispose 取消订阅。

发布事件时的派发逻辑

// LocalEventBus.cs(简化)
public async Task PublishAsync<TEvent>(TEvent eventData, bool onUnitOfWorkComplete = true)

{
    if
 (onUnitOfWorkComplete && _unitOfWorkManager.Current != null)
    {
        // 如果当前有 UOW,事件暂存,UOW 成功提交后再发送

        _unitOfWorkManager.Current.AddOrReplaceLocalEvent(
            new
 UnitOfWorkEventRecord(Guid.NewGuid(), eventData, typeof(TEvent)));
        return
;
    }

    // 没有 UOW,立即发送

    await
 TriggerHandlersAsync(typeof(TEvent), eventData);
}

这就是事件与 UOW 联动的实现:不管调不调 onUnitOfWorkComplete,只要当前有 UOW,事件就暂存到 UOW 中,等 Complete 时再触发。

触发处理器的实现

private async Task TriggerHandlersAsync(Type eventType, object eventData)
{
    var
 handlers = GetHandlerFactories(eventType);  // 获取该事件的所有处理器工厂
    foreach
 (var handlerFactory in handlers)
    {
        using
 var handler = handlerFactory.GetHandler();  // 从 DI 解析处理器
        await
 handler.HandleEventAsync(eventData);         // 执行
    }
}

关键设计:每个处理器使用独立的 ILocalEventHandler 实例,一个处理器失败不影响其他处理器。

一对多:发布一次,多个处理器收到

PublishAsync(new OrderCreatedEvent(...))
  ↓
EventHandlerFactories[OrderCreatedEvent]  → [OrderCreatedHandler, OrderCreatedAnalyticsHandler, ...]
  ├── OrderCreatedHandler.HandleEventAsync()        ← 发通知
  ├── OrderCreatedAnalyticsHandler.HandleEventAsync() ← 统计
  └── ...更多处理器

实战

事务一致性

public async Task CreateOrderAsync(CreateOrderDto input)
{
    var
 order = await _orderRepo.InsertAsync(order);  // 插入数据库
    await
 _localEventBus.PublishAsync(                  // 发布事件
        new
 OrderCreatedEvent(order.Id, ...),
        onUnitOfWorkComplete: true                     // 默认 true,UOW 成功后才发送
    );
    // 如果 UOW 回滚,事件不会被发送

}

处理器隔离

每个处理器相互独立,一个抛出异常不影响其他。但要注意:处理器默认共用同一个 UOW,如果需要在处理器中使用独立的事务,可以在处理器中手动创建:

public class EmailHandler : ILocalEventHandler<OrderCreatedEvent>, ITransientDependency
{
    private
 readonly IUnitOfWorkManager _uowManager;

public async Task HandleEventAsync(OrderCreatedEvent eventData)

    {
        using
 var uow = _uowManager.Begin(new AbpUnitOfWorkOptions
        {
            IsTransactional = true,
            RequiresNew = true
        });
        // 独立的事务

        await
 uow.CompleteAsync();
    }
}

相关学习资料