Netor.EventHub 1.2.6

Netor.EventHub

项目概述

Netor.EventHub 提供一个线程安全的内存事件中心实现，支持根据事件标识 EventID 枚举或自定义字符串订阅、取消订阅以及同步/异步发布实现 IEventArgs 的事件参数。通过 AddEventHub 扩展可将事件中心注册为单例服务，以供 .NET 应用在任意组件中注入使用。

v1.2.4 新特性：源代码生成器新增 IPublisher 强类型扩展方法 Emit/EmitAsync，编译期自动生成，无需手动指定泛型参数。

v1.1.2/v1.1.3 新特性：新增 ISocpSubscriber 接口，支持订阅自动跟踪和批量清理。

v1.1.1 新特性：新增 IEventDispatcher 事件调度中心，支持并发控制、超时处理和背压控制。

v1.1.0 新特性：支持通过 ISocpHub 创建隔离的命名事件中心实例，实现多租户/多模块的订阅隔离。

核心功能

• 多种订阅方式：支持 EventID 枚举、自定义字符串或筛选器订阅，实现精细化监听。
• 异步调度：PublishAsync 使用 EventHandler<T> 委托处理器，适配网络、数据库等异步逻辑。
• 事件上下文：处理器接收 EventHubContext 上下文，包含事件中心引用、事件标识和时间戳。
• 订阅管理：提供 Unsubscribe、HasSubscribers、FindSubscribers，便于实时查询及清理订阅。
• 可扩展事件标识：支持 EventID 枚举和自定义字符串两种方式，按需扩展。
• 订阅隔离：通过 ISocpHub 创建命名实例，实现多租户/多模块的订阅隔离。
• 职责分离：IPublisher 负责发布，ISubscriber 负责订阅，支持单独注入使用。
• 订阅自动跟踪：ISocpSubscriber 自动跟踪订阅，Dispose 时一键清理（v1.1.2 新增）。
• 调度中心：IEventDispatcher 基于 Channel 的高性能异步任务队列，支持并发控制和超时处理（v1.1.1 新增）。
• 源代码生成器：编译期自动为 EventID<T> 子类型生成 ISubscriber.On 和 IPublisher.Emit/EmitAsync 强类型扩展方法（v1.2.4 增强）。

核心类型

EventHandler<TArgs>

事件处理器委托，定义了统一的处理器签名：

public delegate Task<bool> EventHandler<in TArgs>(EventHubContext context, TArgs args) where TArgs : IEventArgs;

EventHubContext

事件上下文结构体（readonly record struct），提供事件处理时的上下文信息：

public readonly record struct EventHubContext(
    IEventHub Hub,        // 事件中心实例引用
    EventID? Event,       // 枚举标识（使用枚举发布时有值）
    string? EventID,      // 字符串标识（始终有值）
    ISocpHub? Socp,       // 关联的隔离事件中心（v1.1.0 新增）
    long Timestamp        // UTC 时间戳（秒）
);

ISocpHub（v1.1.0 新增）

隔离事件中心接口，支持创建命名实例实现订阅隔离：

public interface ISocpHub : IDisposable
{
    Guid Id { get; }              // 当前实例唯一标识
    string Name { get; }          // 实例名称
    IPublisher Publisher { get; } // 发布单元
    ISubscriber Subscriber { get; } // 订阅单元
    ISocpHub Create(string? name = null); // 创建新实例
    void Remove(string eventid, string? argsid = null); // 移除指定订阅
    void Remove(); // 移除所有订阅
}

IPublisher（v1.1.0 新增）

发布者接口，专注于事件发布职责： | 方法 | 说明 | |------|------| | Publish<T>(EventID, payload) | 同步发布（fire-and-forget） | | Publish<T>(string, payload) | 同步发布（字符串标识） | | PublishAsync<T>(EventID, payload) | 异步发布 | | PublishAsync<T>(string, payload) | 异步发布（字符串标识） |

ISubscriber（v1.1.0 新增）

订阅者接口，专注于事件订阅职责： | 方法 | 说明 | |------|------| | Subscribe<T>(EventID, handler) | 使用枚举订阅 | | Subscribe<T>(string, handler) | 使用字符串订阅 | | Subscribe<T>(filter, handler) | 使用筛选器订阅 | | Unsubscribe(Guid) | 根据订阅 ID 取消 | | Unsubscribe(EventID/string, argsid?) | 根据事件标识取消 | | HasSubscribers(EventID/string, argsid?) | 检查订阅者 | | FindSubscribers(EventID/string, argsid?) | 查找订阅者 ID |

ISocpSubscriber（v1.1.2 新增）

支持订阅跟踪的订阅者接口，扩展 ISubscriber：

public interface ISocpSubscriber : ISubscriber, IDisposable
{
    IEnumerable<Guid> Subscriptions { get; }  // 获取所有跟踪的订阅 ID
}

特性：

• 自动跟踪通过此实例创建的所有订阅
• Dispose() 时自动批量取消所有订阅
• 取消订阅时自动从跟踪列表移除
• 线程安全

使用场景：

• ViewModel 生命周期管理：页面销毁时一键清理所有订阅
• 组件级订阅：组件卸载时自动清理
• 避免手动管理多个订阅 ID

IEventArgs

事件参数接口，所有事件参数必须实现此接口：

public interface IEventArgs
{
    string ID { get; set; }  // 事件参数唯一标识，用于筛选
}

IEventHub

事件中心接口，提供完整的订阅/发布 API：

IEventDispatcher（v1.1.1 新增）

事件调度中心接口，基于 Channel 的高性能异步任务队列：

internal interface IEventDispatcher : IDisposable
{
    Task DispatchAsync<T>(EventHubContext context, T payload, 
        IEnumerable<EventHandler<T>?> handlers, CancellationToken cancellationToken = default) where T : IEventArgs;
    int PendingCount { get; }   // 排队等待的任务数
    int ActiveCount { get; }    // 正在执行的任务数
    bool IsRunning { get; }     // 调度器是否运行中
}

EventOptions（v1.1.1 新增）

事件调度器配置选项：

public class EventOptions
{
    public int MaxConcurrency { get; set; } = Environment.ProcessorCount * 2;  // 最大并发数
    public int? QueueCapacity { get; set; } = null;                             // 队列容量（null=无限制）
    public TimeSpan HandlerTimeout { get; set; } = TimeSpan.FromSeconds(30);    // 处理器超时时间
    public bool EnableParallelHandlers { get; set; } = false;                   // 是否并行执行处理器
}

EventID 枚举说明

快速使用

1. 注册服务

// Minimal API / ASP.NET Core
var builder = WebApplication.CreateBuilder(args);
builder.Services.AddEventHub();

// 自定义配置调度器选项（v1.1.1 新增）
builder.Services.AddEventHub(options =>
{
    options.MaxConcurrency = 16;                          // 最大并发数
    options.QueueCapacity = 1000;                         // 队列容量
    options.HandlerTimeout = TimeSpan.FromSeconds(10);    // 超时时间
    options.EnableParallelHandlers = false;               // 顺序执行
});

// .NET MAUI (MauiProgram.cs)
builder.Services.AddEventHub();

2. 定义事件参数

public class MessageEventArgs : IEventArgs
{
    public string From { get; set; } = string.Empty;
    public string To { get; set; } = string.Empty;
    public string Content { get; set; } = string.Empty;
    public string ID { get; set; } = Guid.NewGuid().ToString("N");
}

3. 订阅事件

var hub = app.Services.GetRequiredService<IEventHub>();

// 方式一：使用 EventID 枚举订阅
var subscriptionId = hub.Subscribe<MessageEventArgs>(
    EventID.MessageRecived,
    async (context, args) =>
    {
        Console.WriteLine($"[{context.EventID}] 收到消息: {args.Content}");
        return false; // false 继续传递，true 停止传播
    });

// 方式二：使用自定义字符串订阅（适合动态事件）
hub.Subscribe<MessageEventArgs>(
    "chat.room.123",
    async (context, args) =>
    {
        Console.WriteLine($"聊天室消息: {args.Content}");
        return false;
    });

// 方式三：使用筛选器订阅（区分会话/租户）
hub.Subscribe<MessageEventArgs>(
    filter: info => info.eventid == EventID.Custom && info.argsid == "tenant-A",
    handler: async (context, args) =>
    {
        // 只处理指定 argsid 的事件
        return true; // 停止传播
    });

4. 发布事件

// 使用 EventID 枚举发布
await hub.PublishAsync(
    EventID.MessageRecived,
    new MessageEventArgs
    {
        From = "Alice",
        To = "Bob",
        Content = "Hello from MAUI"
    });

// 使用自定义字符串发布
await hub.PublishAsync(
    "chat.room.123",
    new MessageEventArgs { Content = "房间消息" });

// 同步发布（阻塞当前线程）
hub.Publish(EventID.MessageRecived, new MessageEventArgs { Content = "同步消息" });

5. 管理订阅

// 取消订阅（推荐方式）
hub.Unsubscribe(subscriptionId);

// 按事件标识批量取消
hub.Unsubscribe(EventID.MessageRecived);
hub.Unsubscribe("chat.room.123");

// 检查是否有订阅者（避免无订阅时构造复杂负载）
if (hub.HasSubscribers(EventID.MessageRecived))
{
    await hub.PublishAsync(EventID.MessageRecived, payload);
}

// 查找所有订阅者 ID
var subscribers = hub.FindSubscribers(EventID.Custom, "tenant-A");
foreach (var id in subscribers)
{
    Console.WriteLine($"订阅者: {id}");
}

6. 使用源代码生成器扩展方法（v1.2.4 新增）

定义自定义 EventID<T> 子类型后，源代码生成器会自动生成强类型扩展方法：

using Netor.EventHub;

// 定义自定义事件标识
public record ChatMessageArgs : IEventArgs
{
    public string ID { get; set; } = Guid.NewGuid().ToString("N");
    public string From { get; set; } = string.Empty;
    public string Content { get; set; } = string.Empty;
}

public record ChatMessageEventID() : EventID<ChatMessageArgs>("chat.message");

var subscriber = app.Services.GetRequiredService<ISubscriber>();
var publisher = app.Services.GetRequiredService<IPublisher>();

// 自动生成的 On 扩展方法 — 无需手动指定泛型参数
subscriber.On(new ChatMessageEventID(), async (context, args) =>
{
    Console.WriteLine($"收到来自 {args.From} 的消息: {args.Content}");
    return false;
});

// 自动生成的 Emit 扩展方法 — 同步发布
publisher.Emit(new ChatMessageEventID(), new ChatMessageArgs
{
    From = "Alice",
    Content = "Hello!"
});

// 自动生成的 EmitAsync 扩展方法 — 异步发布
await publisher.EmitAsync(new ChatMessageEventID(), new ChatMessageArgs
{
    From = "Bob",
    Content = "Hi there!"
});

7. 使用隔离事件中心（v1.1.0 新增）

// 注入 ISocpHub 创建隔离实例
var socp = app.Services.GetRequiredService<ISocpHub>();

// 创建命名实例（订阅相互隔离）
var roomHub = socp.Create("chat-room-123");
var systemHub = socp.Create("system-notifications");

// 在隔离实例中订阅（仅接收该实例发布的事件）
roomHub.Subscriber.Subscribe<MessageEventArgs>(
    EventID.MessageRecived,
    async (context, args) =>
    {
        Console.WriteLine($"[房间消息] {args.Content}");
        return false;
    });

// 在隔离实例中发布（仅该实例的订阅者收到）
await roomHub.Publisher.PublishAsync(
    EventID.MessageRecived,
    new MessageEventArgs { Content = "仅房间内可见" });

// 销毁实例时自动清理所有订阅
roomHub.Dispose();

8. 单独使用 IPublisher / ISubscriber

// 只需要发布功能时，注入 IPublisher
public class NotificationService(IPublisher publisher)
{
    public async Task SendAsync(string message)
    {
        await publisher.PublishAsync(
            EventID.Broadcast,
            new BroadcastEventArgs { Message = message });
    }
}

// 只需要订阅功能时，注入 ISubscriber
public class MessageHandler(ISubscriber subscriber)
{
    private Guid _subscriptionId;

    public void Start()
    {
        _subscriptionId = subscriber.Subscribe<MessageEventArgs>(
            EventID.MessageRecived,
            async (context, args) =>
            {
                await ProcessMessageAsync(args);
                return false;
            });
    }

    public void Stop() => subscriber.Unsubscribe(_subscriptionId);
}

9. 使用 ISocpSubscriber 自动跟踪订阅（v1.1.2 新增）

using Netor.EventHub.Interfances;

// 注入 ISocpSubscriber（需要先注册）
public class ChatViewModel : IDisposable
{
    private readonly ISocpSubscriber _subscriber;

    public ChatViewModel(ISocpSubscriber subscriber)
    {
        _subscriber = subscriber;

        // 创建多个订阅，无需手动保存每个 ID
        _subscriber.Subscribe<MessageEventArgs>(
            EventID.MessageRecived,
            async (context, args) =>
            {
                Console.WriteLine($"收到消息: {args.Content}");
                return false;
            });

        _subscriber.Subscribe<BroadcastEventArgs>(
            EventID.Broadcast,
            async (context, args) =>
            {
                Console.WriteLine($"广播: {args.Message}");
                return false;
            });

        _subscriber.Subscribe<MessageEventArgs>(
            "chat.room.123",
            async (context, args) =>
            {
                Console.WriteLine($"房间消息: {args.Content}");
                return false;
            });

        // 可以查看当前跟踪的订阅数量
        Console.WriteLine($"当前订阅数: {_subscriber.Subscriptions.Count()}");
    }

    public void Dispose()
    {
        // 一键清理所有订阅！
        _subscriber.Dispose();
    }
}

对比传统方式：

// ❌ 传统方式：需要手动管理每个订阅 ID
private readonly List<Guid> _subscriptionIds = [];

public void Start()
{
    _subscriptionIds.Add(subscriber.Subscribe(...));
    _subscriptionIds.Add(subscriber.Subscribe(...));
    _subscriptionIds.Add(subscriber.Subscribe(...));
}

public void Stop()
{
    foreach (var id in _subscriptionIds)
        subscriber.Unsubscribe(id);
    _subscriptionIds.Clear();
}

// ✅ ISocpSubscriber 方式：自动跟踪，一键清理
public void Dispose() => _subscriber.Dispose();

完整示例

Samples/EventHub/Program.cs 展示了 Minimal API 主机中集成事件中心的完整流程：

using Netor.EventHub;

var builder = WebApplication.CreateBuilder(args);
builder.Services.AddEventHub();

var app = builder.Build();
var hub = app.Services.GetRequiredService<IEventHub>();

// 注册多个订阅者（链式处理）
hub.Subscribe<MessageEventArgs>(
    EventID.MessageRecived,
    async (context, msg) =>
    {
        Console.ForegroundColor = ConsoleColor.Green;
        Console.WriteLine($"New message {msg.Content}");
        Console.ResetColor();
        return false; // 继续传播
    });

hub.Subscribe<MessageEventArgs>(
    EventID.MessageRecived,
    async (context, msg) =>
    {
        Console.ForegroundColor = ConsoleColor.Yellow;
        Console.WriteLine($"{DateTime.Now:HH:mm:ss} message {msg.Content}");
        Console.ResetColor();
        return false;
    });

hub.Subscribe<BroadcastEventArgs>(
    EventID.Broadcast,
    async (context, msg) =>
    {
        Console.ForegroundColor = ConsoleColor.Blue;
        Console.WriteLine($"Broadcast message: {msg.Message}");
        Console.ResetColor();
        return false;
    });

// 后台任务发布事件
_ = Task.Run(async () =>
{
    await Task.Delay(2000);
    await hub.PublishAsync(EventID.MessageRecived, new MessageEventArgs { Content = "Hello" });
    await Task.Delay(2000);
    await hub.PublishAsync(EventID.Broadcast, new BroadcastEventArgs { Message = "系统通知" });
});

Console.ReadLine();

public class MessageEventArgs : IEventArgs
{
    public string From { get; set; } = Guid.NewGuid().ToString("N");
    public string To { get; set; } = Guid.NewGuid().ToString("N");
    public string Content { get; set; } = "this is a message content";
    public string ID { get; set; } = Guid.NewGuid().ToString("N");
}

public class BroadcastEventArgs : IEventArgs
{
    public string Message { get; set; } = "this is a broadcast message";
    public string ID { get; set; } = Guid.NewGuid().ToString("N");
}

.NET MAUI 集成示例

// MauiProgram.cs
public static MauiApp CreateMauiApp()
{
    var builder = MauiApp.CreateBuilder();
    builder.UseMauiApp<App>();
    builder.Services.AddEventHub();
    return builder.Build();
}

// ViewModel 中使用（传统方式）
public partial class ChatViewModel : ObservableObject
{
    private readonly IEventHub _hub;
    private Guid _subscriptionId;

    public ChatViewModel(IEventHub hub)
    {
        _hub = hub;
        _subscriptionId = _hub.Subscribe<MessageEventArgs>(
            EventID.MessageRecived,
            async (context, args) =>
            {
                // 切换到 UI 线程更新界面
                MainThread.BeginInvokeOnMainThread(() =>
                {
                    Messages.Add(args);
                });
                return false;
            });
    }

    public void Cleanup()
    {
        _hub.Unsubscribe(_subscriptionId);
    }
}

// ViewModel 中使用（推荐：使用 ISocpSubscriber，v1.1.2+）
public partial class ChatViewModel : ObservableObject, IDisposable
{
    private readonly ISocpSubscriber _subscriber;

    public ChatViewModel(ISocpSubscriber subscriber)
    {
        _subscriber = subscriber;

        // 订阅多个事件，无需手动保存 ID
        _subscriber.Subscribe<MessageEventArgs>(
            EventID.MessageRecived,
            async (context, args) =>
            {
                MainThread.BeginInvokeOnMainThread(() => Messages.Add(args));
                return false;
            });

        _subscriber.Subscribe<BroadcastEventArgs>(
            EventID.Broadcast,
            async (context, args) =>
            {
                MainThread.BeginInvokeOnMainThread(() => Notifications.Add(args));
                return false;
            });
    }

    // 页面销毁时自动清理所有订阅
    public void Dispose() => _subscriber.Dispose();
}

项目结构

Netor.EventHub/
├── Src/Netor.EventHub/
│   ├── Services/
│   │   ├── EventHub.cs          # IEventHub 门面实现
│   │   ├── Publisher.cs         # IPublisher 实现
│   │   ├── Subscriber.cs        # ISubscriber 实现
│   │   ├── SocpSubscriber.cs    # ISocpSubscriber 实现（v1.1.2）
│   │   ├── SocpHub.cs           # ISocpHub 实现
│   │   ├── EventStore.cs        # 共享订阅存储
│   │   └── EventHubContext.cs   # 事件上下文
│   ├── Interfances/
│   │   ├── IEventHub.cs         # 事件中心接口
│   │   ├── IPublisher.cs        # 发布者接口
│   │   ├── ISubscriber.cs       # 订阅者接口
│   │   ├── ISocpSubscriber.cs   # 订阅跟踪接口（v1.1.2）
│   │   ├── ISocpHub.cs          # 隔离事件中心接口
│   │   └── IEventDispatcher.cs  # 事件调度中心接口（v1.1.1）
│   ├── EventHandler~.cs         # 事件处理器委托
│   ├── EventID.cs               # 事件标识枚举
│   ├── EventOptions.cs          # 调度器配置选项（v1.1.1）
│   ├── IEventArgs.cs            # 事件参数接口
│   └── Extensions/
│       └── ServiceCollectionExtensions.cs  # DI 扩展
├── Src/Netor.EventHub.Generator/
│   ├── SubscriberExtensionGenerator.cs  # 增量源代码生成器（生成 ISubscriber.On + IPublisher.Emit/EmitAsync）
│   └── EventIdTypeInfo.cs               # EventID<T> 元数据结构体
└── Samples/EventHub.Samples/
    └── Program.cs               # 示例程序

最佳实践

架构选择

• 简单场景：直接注入 IEventHub，统一管理订阅和发布
• 职责分离：注入 IPublisher 或 ISubscriber，按需使用
• 多租户/模块隔离：使用 ISocpHub.Create() 创建命名实例

性能优化

• 使用 HasSubscribers 在关键路径前做预检，避免无订阅场景下构造复杂负载
• 优先使用 Unsubscribe(Guid) 精确取消订阅，性能优于按事件标识批量取消
• 长耗时操作使用 PublishAsync，避免阻塞调用线程

调度器配置（v1.1.1 新增）

根据应用场景调整 EventOptions：

builder.Services.AddEventHub(options =>
{
    // I/O 密集型场景：提高并发数
    options.MaxConcurrency = Environment.ProcessorCount * 4;

    // 高负载场景：启用背压控制
    options.QueueCapacity = 5000;

    // 快速响应场景：缩短超时时间
    options.HandlerTimeout = TimeSpan.FromSeconds(5);

    // 无依赖处理器：启用并行执行
    options.EnableParallelHandlers = true;
});

事件设计

• 将自定义 EventID 与业务模块对应，配合 argsid 传递会话或租户信息
• 使用字符串事件标识实现动态事件（如 chat.room.{roomId}）
• 事件参数类实现 IEventArgs，ID 属性用于筛选分发

MAUI 注意事项

• 在处理器中使用 MainThread.BeginInvokeOnMainThread 切换到 UI 线程
• 在 ViewModel 销毁时调用 Unsubscribe 清理订阅，避免内存泄漏
• 考虑在 OnDisappearing 或 IDisposable.Dispose 中取消订阅
• 使用 ISocpHub 为每个页面创建隔离实例，页面销毁时自动清理

异常处理

• 处理器中抛出的异常会被捕获并通过 ILogger 记录，不会向上传播
• 异常不会中断后续订阅者的执行
• 处理器超时会记录警告日志，并继续执行下一个处理器（v1.1.1）

ISocpHub 使用场景

• 多租户应用：为每个租户创建独立实例，事件相互隔离
• 模块化架构：各模块使用独立实例，避免事件冲突
• 聊天室/频道：每个房间一个实例，成员离开时自动清理
• 页面级事件：MAUI 页面使用独立实例，页面销毁时一键清理

许可证

MIT License