跳至主要內容

28. 事件总线实战

后端手册事件总线异步大约 9 分钟约 2673 字

一句话定位：事件总线是 Furion 的进程内/跨进程发布订阅机制，Admin.NET 用它承担所有"主流程不该等"的活儿——写日志、发邮件、发短信、推企微、推在线用户——让请求线程尽快返回。

一、架构与开关

业务代码 / 框架内部
   │ IEventPublisher.PublishAsync(eventId, payload)
   ▼
┌──────────────────────────────────────────────────┐
│ IEventSourceStorer（事件源存储器）                 │
│   Memory（默认）                                   │
│   Redis    → RedisEventSourceStorer                │
│   RabbitMQ → RabbitMQEventSourceStore              │
│   队列名：adminnet_eventsource_queue               │
└───────────────┬──────────────────────────────────┘
                │ 后台分发
                ▼
      RetryEventHandlerExecutor（失败重试包装器）
                │
                ▼
      IEventSubscriber 上带 [EventSubscribe(eventId)] 的方法

开关配置：

// Admin.NET.Application/Configuration/MessageQueue.json
"EventBus": {
  "EventSourceType": "Memory"     // Memory、Redis、RabbitMQ、Kafka
}

注册在 Admin.NET.Web.Core/Startup.cs：

services.AddEventBus(options =>
{
    options.UseUtcTimestamp = false;
    options.LogEnabled = false;                                  // 关闭 Furion 事件日志
    options.AddExecutor<RetryEventHandlerExecutor>();            // 失败重试执行器
    options.UnobservedTaskExceptionHandler = (obj, args) =>
    {
        if (args.Exception?.Message != null)
            Log.Error($"事件总线未处理异常 ：{args.Exception?.Message} ", args.Exception);
    };
    // options.AddMonitor<EventHandlerMonitor>();                 // 可选监视器（已实现但默认关闭）

    var eventSourceType = App.GetConfig<string>("EventBus:EventSourceType", true);
    if (eventSourceType == "Redis") { /* 且 Cache.CacheType == Redis 才生效 */ }
    else if (eventSourceType == "RabbitMQ") { /* 用 RabbitMqOptions 建连接 */ }
});

options.AddExecutor<RetryEventHandlerExecutor>() 是点睛之笔：默认只有把执行器注册进去，[EventSubscribe(NumRetries = n, RetryTimeout = ms)] 才会真正生效。

二、发布（Publish）

注入 Furion 的 IEventPublisher，三种重载：

public class BizOrderService : IDynamicApiController, ITransient
{
    private readonly IEventPublisher _eventPublisher;

    public BizOrderService(IEventPublisher eventPublisher) => _eventPublisher = eventPublisher;

    [DisplayName("提交订单")]
    public async Task Submit(SubmitOrderInput input)
    {
        // ① 带 Payload 的具名事件（最常用）
        await _eventPublisher.PublishAsync(nameof(OrderCreatedEvent),
            new OrderCreatedEvent { OrderId = order.Id, OrderNo = order.OrderNo });

        // ② 封装成 ChannelEventSource（可指定 Channel 与 Source）
        await _eventPublisher.PublishAsync(new ChannelEventSource(nameof(OrderPaidEvent), payload));

        // ③ 纯通知（无 Payload）
        await _eventPublisher.PublishAsync("CacheCleared");
    }
}

框架内部现成的例子：

// JobMonitor.OnExecutedAsync：定时任务失败走事件总线发邮件 + 落异常日志
await _eventPublisher.PublishAsync(nameof(AppEventSubscriber.SendErrorMail), exception, stoppingToken);
await _eventPublisher.PublishAsync(new ChannelEventSource(nameof(AppEventSubscriber.CreateExLog), sysLogEx));

// SysAuthService.Logout：发布用户退出事件
await _eventPublisher.PublishAsync(UserEventTypeEnum.Logout, _userManager);

三、订阅（Subscribe）

Admin.NET.Core/EventBus/AppEventSubscriber.cs 是框架自带的订阅器，写法：

public class AppEventSubscriber : IEventSubscriber, ISingleton, IDisposable
{
    private readonly IServiceScope _serviceScope;
    private readonly SysConfigService _sysConfigService;
    private readonly SysWeComService _sysWeComService;

    public AppEventSubscriber(IServiceScopeFactory scopeFactory)
    {
        _serviceScope = scopeFactory.CreateScope();
        _sysConfigService = _serviceScope.ServiceProvider.GetRequiredService<SysConfigService>();
        _sysWeComService  = _serviceScope.ServiceProvider.GetRequiredService<SysWeComService>();
    }

    [SuppressMonitor]
    [EventSubscribe(nameof(CreateHttpLog))]
    public async Task CreateHttpLog(EventHandlerExecutingContext<SysLogHttp> context)
    {
        // CopyNew() 获取独立连接，避免事件总线后台线程与请求线程共享作用域连接导致偶发空引用
        await SqlSugarSetup.LogDb.CopyNew().Insertable(context.Payload).ExecuteCommandAsync();
    }

    [SuppressMonitor]
    [EventSubscribe(nameof(SendErrorMail))]
    public async Task SendErrorMail(EventHandlerExecutingContext context)
    {
        if (!await _sysConfigService.GetConfigValueByCode<bool>(ConfigConst.SysErrorMail)) return;
        var title = $"{await _sysConfigService.GetConfigValueByCode<string>(ConfigConst.SysWebTitle)} 系统异常";
        await _serviceScope.ServiceProvider.GetRequiredService<SysEmailService>()
            .SendEmail(JSON.Serialize(SerializableException.FromException(context.GetPayload<Exception>())), title);
    }

    public void Dispose() => _serviceScope.Dispose();
}

要点：

要点说明
类标记IEventSubscriber + ISingleton（订阅器常驻，别用 Scoped）
方法签名Task 方法名(EventHandlerExecutingContext<T> context) 或 EventHandlerExecutingContext
取数据强类型版直接 context.Payload；弱类型版用 context.GetPayload<T>()
[SuppressMonitor]跳过 LoggingMonitor，避免写日志的动作再被审计一次而产生递归日志

自定义事件：完整可运行例子

namespace Admin.NET.Application;

/// <summary>
/// 订单已创建事件（Payload）
/// </summary>
public class OrderCreatedEvent
{
    public long OrderId { get; set; }
    public string OrderNo { get; set; }
    public decimal Amount { get; set; }
}
/// <summary>
/// 业务事件订阅器
/// </summary>
public class BizEventSubscriber : IEventSubscriber, ISingleton, IDisposable
{
    private readonly IServiceScopeFactory _scopeFactory;
    private readonly ILogger<BizEventSubscriber> _logger;

    public BizEventSubscriber(IServiceScopeFactory scopeFactory, ILoggerFactory loggerFactory)
    {
        _scopeFactory = scopeFactory;
        _logger = loggerFactory.CreateLogger<BizEventSubscriber>();
    }

    /// <summary>
    /// 订单创建后：记出库流水 + 通知仓管
    /// </summary>
    [SuppressMonitor]
    [EventSubscribe(nameof(OrderCreatedEvent), NumRetries = 3, RetryTimeout = 2000)]
    public async Task OnOrderCreated(EventHandlerExecutingContext<OrderCreatedEvent> context)
    {
        var payload = context.Payload;

        using var scope = _scopeFactory.CreateScope();
        // ⚠️ 后台线程统一：CreateScope + CopyNew()
        var db = scope.ServiceProvider.GetRequiredService<ISqlSugarClient>().CopyNew();

        await db.Insertable(new BizStockFlow
        {
            BizId = payload.OrderId, BizNo = payload.OrderNo,
            Amount = payload.Amount, CreateTime = DateTime.Now
        }).ExecuteCommandAsync();

        _logger.LogInformation("订单 {OrderNo} 已生成出库流水", payload.OrderNo);
    }

    public void Dispose() { }
}
// 发布端
await _eventPublisher.PublishAsync(nameof(OrderCreatedEvent), new OrderCreatedEvent
{
    OrderId = 123456789, OrderNo = "SO20261018001", Amount = 998.00m
});

四、事件 Id 约定

框架内部的约定是 nameof(方法名) 即事件 Id，好处是编译期可追踪、重构安全：

事件 Id订阅方法作用
CreateHttpLogCreateHttpLog落 HTTP 请求日志（SysLogHttp）
CreateExLogCreateExLog落异常日志（SysLogEx）
CreateMsgLogCreateMsgLog落消息日志（SysLogMsg）
SendErrorMailSendErrorMail异常邮件（受 SysErrorMail 系统参数控制）
SendErrorWechatSendErrorWechat异常推企业微信（受 SysErrorWeCom 控制）
SendSmsSendSms发短信（SendSmsInput）
UserEventTypeEnum.Logout—用户退出（枚举值做 Id）

自定义事件建议：

  • 用 nameof(XxxEvent) 或专门的常量类，避免魔法字符串
  • 语义用过去式：OrderCreatedEvent / UserRegisteredEvent
  • 一个事件可被多个订阅方法同时订阅（多点投递），反之一个方法订阅多个 Id 需写多个 [EventSubscribe]

五、内置事件速查（你在代码里能直接遇到的）

请求进入 → LoggingMonitor / 异常处理
    ├─► CreateHttpLog → SysLogHttp 落库（SqlSugarSetup.LogDb）
    ├─► CreateExLog   → SysLogEx 落库
    ├─► SendErrorMail → 读 ConfigConst.SysErrorMail 开关 → SysEmailService.SendEmail
    └─► SendErrorWechat → 读 ConfigConst.SysErrorWeCom 开关 → SysWeComService.PushErrorMessage

定时任务异常 → JobMonitor.OnExecutedAsync
    ├─► SendErrorMail（同上）
    └─► CreateExLog（同上，并按 SysErrorWeCom 推企微 scheduled task 消息）

发短信 → SendSms(SendSmsInput)
消息中心 → CreateMsgLog(SysLogMsg)

SqlSugarSetup.LogDb 的定义：

public static SqlSugarScopeProvider LogDb => ITenant.IsAnyConnection(SqlSugarConst.LogConfigId)
    ? ITenant.GetConnectionScope(SqlSugarConst.LogConfigId)
    : ITenant.GetConnectionScope(SqlSugarConst.MainConfigId);

即：配了独立的日志库（LogConfigId）就写日志库，否则写主库。

六、Redis / RabbitMQ 事件源切换

Memory（默认）

  • 事件存在进程内存队列
  • 进程重启 = 事件丢失，且多实例之间互不可见
  • 适合：开发环境、对可靠性无要求的旁路逻辑

Redis

"Cache": { "CacheType": "Redis", "Redis": { "ConnectionString": "127.0.0.1:6379,password=,defaultDatabase=1" } },
"EventBus": { "EventSourceType": "Redis" }
if (eventSourceType == "Redis")
{
    if (App.GetConfig<CacheOptions>("Cache", true).CacheType == CacheTypeEnum.Redis.ToString())
    {
        options.ReplaceStorerOrFallback(serviceProvider =>
        {
            var cacheProvider = serviceProvider.GetRequiredService<NewLife.Caching.ICacheProvider>();
            return new RedisEventSourceStorer(cacheProvider, "adminnet_eventsource_queue", 3000);
        });
    }
}

隐藏前提：Cache.CacheType 必须已经是 Redis，否则这段判断不成立、静默回落到 Memory。切换前先把缓存改成 Redis 并验证。 第三个参数 3000 是轮询/拉取间隔（毫秒）。

RabbitMQ

"EventBus": { "EventSourceType": "RabbitMQ" },
"RabbitMQ": { "UserName": "adminnet", "Password": "adminnet++123456",
              "HostName": "127.0.0.1", "Port": 5672, "VirtualHost": "/" }
else if (eventSourceType == "RabbitMQ")
{
    var o = App.GetConfig<RabbitMqOptions>("RabbitMQ");
    var storer = new RabbitMQEventSourceStore(new ConnectionFactory
    {
        UserName = o.UserName, Password = o.Password,
        HostName = o.HostName, Port = o.Port, VirtualHost = o.VirtualHost,
    }, "adminnet_eventsource_queue", 3000);
    options.ReplaceStorerOrFallback(_ => storer);
}

选型对照表

场景推荐理由
本地开发Memory零依赖
单实例生产Memory / Redis事件丢一点无所谓就用 Memory
多实例集群Redis 或 RabbitMQ保证每个实例都能消费到
日志绝不丢RabbitMQ持久化 + ACK
已有 Redis 缓存Redis复用中间件，成本最低

注意：订阅者仍在各应用进程内。换了存储器只解决"事件不丢、能被别的实例看到"，不解决"没有实例在线时事件自动执行"。

七、失败重试 RetryEventHandlerExecutor

// Admin.NET.Core/EventBus/RetryEventHandlerExecutor.cs
public class RetryEventHandlerExecutor : IEventHandlerExecutor
{
    public async Task ExecuteAsync(EventHandlerExecutingContext context,
                                   Func<EventHandlerExecutingContext, Task> handler,
                                   CancellationToken cancellationToken)
    {
        var eventSubscribeAttribute = context.Attribute;
        await Retry.InvokeAsync(
            async () => await handler(context),
            eventSubscribeAttribute?.NumRetries ?? 0,       // 重试次数，默认 0
            eventSubscribeAttribute?.RetryTimeout ?? 1000,  // 重试间隔（毫秒），默认 1000
            exceptionTypes: eventSubscribeAttribute?.ExceptionTypes,
            retryAction: (total, times) =>
                _logger.LogWarning($"事件处理器 {context.Source.EventId} 第 {times}/{total} 次重试"));
    }
}

用法：

// 只重试指定异常类型，最多 3 次，间隔 2 秒
[EventSubscribe(nameof(OrderCreatedEvent),
    NumRetries = 3,
    RetryTimeout = 2000,
    ExceptionTypes = new[] { typeof(TimeoutException), typeof(InvalidOperationException) })]
public async Task OnOrderCreated(EventHandlerExecutingContext<OrderCreatedEvent> context) { ... }

参数语义：

参数默认说明
NumRetries00 表示不重试（首次执行算 1 次，重试 N 次）
RetryTimeout1000两次尝试之间的等待毫秒数
ExceptionTypesnull（全部异常）只有这些类型才重试
FallbackPolicynull降级策略（源码中有注释掉的 IEventFallbackPolicy 用法）

还有一种辅助设施：EventHandlerMonitor（Admin.NET.Core/EventBus/EventHandlerMonitor.cs），实现了 IEventHandlerMonitor，可在执行前/后/异常时埋点，默认被注释掉，需要时打开：

options.AddMonitor<EventHandlerMonitor>();

八、⚠️ 后台线程使用仓储的坑（CopyNew）

这是 Admin.NET 事件总线里最常见、也最容易被忽视的崩溃源。

根因：ISqlSugarClient 是 单例（services.AddSingleton<ISqlSugarClient>(sqlSugar)），而 SqlSugarRepository<T> 是 Scoped。事件总线的投递线程不属于任何 HTTP 请求作用域，于是：

  • 在已过生命周期的 Scoped 实例上调用方法 → ObjectDisposedException: Cannot access a disposed object
  • 或者：多个线程并发复用同一个连接 → SqlSugar 报 "Connection is busy" / 偶发空引用
  • 或者：因为 App.User 为 null，所有全局过滤器（租户/软删除/数据范围）都不生效（见第 13 篇）

框架的官方解法（源码注释原文）：

// AppEventSubscriber.cs —— 每个日志事件方法里的同一行注释
// CopyNew() 获取独立连接，避免事件总线后台线程与请求线程共享作用域连接导致偶发空引用
await SqlSugarSetup.LogDb.CopyNew().Insertable(context.Payload).ExecuteCommandAsync();

三种正确写法：

// ① 直接用静态的 LogDb + CopyNew（框架日志类事件的标准写法）
await SqlSugarSetup.LogDb.CopyNew().Insertable(log).ExecuteCommandAsync();

// ② 自己建 Scope，取 ISqlSugarClient 后 CopyNew（业务事件推荐）
using var scope = _scopeFactory.CreateScope();
var db = scope.ServiceProvider.GetRequiredService<ISqlSugarClient>().CopyNew();
await db.Insertable(entity).ExecuteCommandAsync();

// ③ 取仓储后 CopyNew（CleanSysLogJob 的写法）
var rep = serviceScope.ServiceProvider.GetRequiredService<SqlSugarRepository<SysLogVis>>().CopyNew();
await rep.AsDeleteable().Where(...).ExecuteCommandAsync();

错误写法（会炸）：

public class BadSubscriber : IEventSubscriber, ISingleton
{
    private readonly SqlSugarRepository<SysLogOp> _rep;   // ❌ Scoped 注入进 Singleton

    [EventSubscribe(nameof(CreateOpLog))]
    public async Task CreateOpLog(EventHandlerExecutingContext context)
    {
        await _rep.InsertAsync(...);                       // ❌ 作用域已释放 / 跨线程争用
    }
}

记住三字诀：Singleton 订阅器 + CreateScope + CopyNew。这也同样适用于第 16 篇的 MQTT 拦截器、第 17 篇的 RabbitMQ handler、第 19 篇的定时任务。

另一个隐坑：订阅器里读不到 App.User。如果需要按租户/用户处理，请在发布端把 UserManager 的 UserId/TenantId 塞进 Payload，别指望订阅端解析 Claims。

九、步骤清单

  1. 定义 Payload 类（POCO，别带导航属性）
  2. 写 XxxEventSubscriber : IEventSubscriber, ISingleton, IDisposable
  3. 订阅方法贴 [EventSubscribe(nameof(XxxEvent))] + [SuppressMonitor]，按需配 NumRetries/RetryTimeout
  4. 方法内部：CreateScope() → GetRequiredService<ISqlSugarClient>().CopyNew()
  5. 需要的用户信息在发布端塞进 Payload
  6. 发布端注入 IEventPublisher，PublishAsync(nameof(XxxEvent), payload)
  7. 生产环境按集群需求切换 EventBus.EventSourceType
  8. 观察是否出现 事件总线未处理异常 日志（UnobservedTaskExceptionHandler 兜底）

⚠️ 注意事项 / 常见坑

  1. ISingleton 订阅器千万别注入 Scoped 服务（UserManager、SqlSugarRepository<T>、SysCacheService 若注册为 Scoped）。必须用 IServiceScopeFactory.CreateScope()。
  2. 未注册 AddExecutor 时 [EventSubscribe(NumRetries=n)] 无效：只有把 RetryEventHandlerExecutor 加进 options 才会走重试逻辑。
  3. NumRetries = 0 是默认值：不加参数不会有自动重试，事件失败就只留一条 UnobservedTaskExceptionHandler 的日志。
  4. Redis 模式的前置条件：Cache.CacheType 必须同时为 Redis，否则 ReplaceStorerOrFallback 根本不执行，你会以为"配了 Redis 但还是内存"。
  5. 事件不是分布式任务调度：没有实例在线时事件无人消费；Memory 模式重启即丢。
  6. 订阅方法与发布是同进程的：跨服务请用 RabbitMQ 业务队列（第 17 篇）而不是事件总线。
  7. [SuppressMonitor] 别漏：少了它，写日志的事件本身又会触发 LoggingMonitor → 再写日志 → 递归。
  8. App.User 为 null 导致过滤器失效：后台线程写入的数据不会被自动打 TenantId/CreateUserId；要手动赋值。
  9. ISingleton 订阅器记得 Dispose() 释放自己持有的 Scope（框架里 AppEventSubscriber.Dispose() 就是干这个的）。
  10. 消息体要可序列化：PublishAsync 的 Payload 会被序列化进 Redis/RabbitMQ，别塞 IFormFile、DbConnection 这类对象。
  11. UseUtcTimestamp = false：时间戳按本地时间，跨时区集群要注意。

小结

  • 发布：IEventPublisher.PublishAsync(eventId, payload)；订阅：[EventSubscribe(eventId)] + IEventSubscriber
  • 事件 Id 用 nameof(XxxEvent) / nameof(方法名)，框架内置的六个见第四节表格
  • 存储器三选一：Memory（默认，丢）/ Redis（需 CacheType 也是 Redis）/ RabbitMQ（最可靠），队列名 adminnet_eventsource_queue
  • 重试靠 RetryEventHandlerExecutor + [EventSubscribe(NumRetries, RetryTimeout, ExceptionTypes)]
  • 后台线程的铁律：Singleton 订阅器 + CreateScope() + CopyNew()，且 App.User 为空，所需用户信息要在发布端塞进 Payload