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();
    }
}