.NET类库场景下用async/await替代传统事件的方案问询

场景背景 #

假设有一个面向控制台、类库或UI应用的类库，支持向远程端点发送命令并接收响应，同时能接收远程端点的事件消息，用户还可响应远程事件发送命令。传输机制采用async/await（比如WebSocket）。

传统同步API示例 #

传统.NET应用中的API设计如下（示例代码仅作讨论，不保证可编译）：

private Command currentCommand;

public event EventHandler<EventReceivedEventArgs>? EventReceived;

public CommandResponse ExecuteCommand(CommandParameters parameters)
{
    currentCommand = transport.SendData(parameters.Serialize());
    return currentCommand.WaitForCommandResponse();
}

protected void OnEventReceived(EventData eventData)
{
    if (this.EventReceived is not null)
    {
        this.EventReceived(this, new EventReceivedEventArgs(eventData));
    }
}

private void DataReceiver()
{
    // 示例仅作演示，实际应用会有终止条件
    while (true)
    {
        // 假设此API从远程端获取数据，解析逻辑视为实现细节
        byte[] receivedData = transport.ReceiveData();
        ParsedData parsed = Parse(receivedData);
        if (parsed is CommandResponse)
        {
            this.currentCommand.SetResponse(parsed);
        }

        if (parsed is EventData)
        {
            this.OnEventReceived(parsed);
        }
    }
}

存在问题的异步API示例 #

考虑到.NET事件不适用于async/await、永远不要使用async void、不要混合同步和异步代码这些社区建议，尝试的异步API设计存在明显弊端（仅作讨论占位）：

private Command currentCommand;

// 此处的事件设计被认为是糟糕的API选择
public event EventHandler<EventReceivedEventArgs>? EventReceived;

public async Task<CommandResponse> ExecuteCommandAsync(CommandParameters parameters)
{
    await transport.SendCommandAsync(parameters.Serialize());
    return await transport.WaitForCommandResponse();
}

protected void OnEventReceived(EventData eventData)
{
    if (this.EventReceived is not null)
    {
        this.EventReceived(this, new EventReceivedEventArgs(eventData));
    }
}

private async Task DataReceiver()
{
    while (true)
    {
        await byte[] receivedData = transport.ReceiveData();
        ParsedData parsed = Parse(receivedData);
        if (parsed is CommandResponse)
        {
            this.currentCommand.SetResponse(parsed);
        }

        if (parsed is EventData)
        {
            this.OnEventReceived(parsed);
        }
    }

    return Task.CompletedTask;
}

这种设计的核心问题在于：消费者的事件处理程序会阻塞生产者方法，可能引发挂起、长时间运行或异常等问题。

社区推荐的替代方案 #

针对异步场景下的事件通知需求，.NET社区主要推荐以下几种替代方式：

1. 使用异步事件委托（Func<object, TEventArgs, Task>） #

放弃传统的EventHandler<T>，改用支持异步的委托类型，允许事件处理程序返回Task，生产者可以await所有处理程序完成，避免阻塞或同步调用带来的问题。

示例改造：

// 定义异步事件委托
public event Func<object, EventReceivedEventArgs, Task>? EventReceivedAsync;

protected async Task OnEventReceivedAsync(EventData eventData)
{
    var handler = EventReceivedAsync;
    if (handler != null)
    {
        // 获取所有注册的处理程序并逐个await
        var handlers = handler.GetInvocationList().Cast<Func<object, EventReceivedEventArgs, Task>>();
        foreach (var h in handlers)
        {
            await h(this, new EventReceivedEventArgs(eventData));
        }
    }
}

// 在DataReceiver中调用异步版本
private async Task DataReceiver()
{
    while (true)
    {
        byte[] receivedData = await transport.ReceiveDataAsync();
        ParsedData parsed = Parse(receivedData);
        if (parsed is EventData eventData)
        {
            await OnEventReceivedAsync(eventData);
        }
    }
}

这种方式保留了事件的订阅模式，同时原生支持异步处理，避免了async void的风险。

2. 使用可观察序列（Rx.NET） #

Rx.NET（Reactive Extensions）提供了强大的异步事件流处理能力，将事件转换为可观察的序列，消费者可以通过LINQ操作符处理事件流，天然支持异步操作。

示例思路：

// 定义可观察的事件序列
private readonly Subject<EventData> _eventSubject = new Subject<EventData>();
public IObservable<EventData> Events => _eventSubject.AsObservable();

// 在DataReceiver中推送事件
private async Task DataReceiver()
{
    while (true)
    {
        byte[] receivedData = await transport.ReceiveDataAsync();
        ParsedData parsed = Parse(receivedData);
        if (parsed is EventData eventData)
        {
            _eventSubject.OnNext(eventData);
        }
    }
}

消费者可以这样订阅：

// 异步处理事件流
var subscription = client.Events
    .SelectMany(async e => await ProcessEventAsync(e))
    .Subscribe();

Rx.NET适合复杂的事件流场景，比如过滤、节流、合并等操作，同时完美兼容async/await。

3. 使用通道（Channel） #

.NET Core 3.0+引入的Channel<T>提供了异步的生产者-消费者队列，生产者将事件写入通道，消费者从通道异步读取并处理，完全解耦生产者和消费者的执行流程。

示例思路：

private readonly Channel<EventData> _eventChannel = Channel.CreateUnbounded<EventData>();
public ChannelReader<EventData> EventReader => _eventChannel.Reader;

// 在DataReceiver中写入事件
private async Task DataReceiver()
{
    while (true)
    {
        byte[] receivedData = await transport.ReceiveDataAsync();
        ParsedData parsed = Parse(receivedData);
        if (parsed is EventData eventData)
        {
            await _eventChannel.Writer.WriteAsync(eventData);
        }
    }
}

消费者可以这样处理：

// 异步循环读取事件
await foreach (var eventData in client.EventReader.ReadAllAsync())
{
    await ProcessEventAsync(eventData);
}

通道的优势是轻量级，内置背压机制，适合高吞吐量的异步场景，完全避免了事件订阅的阻塞问题。

4. 回调函数模式 #

如果场景简单，也可以采用注册异步回调函数的方式，让消费者提供处理事件的异步方法，生产者在收到事件时调用并await。

示例思路：

private List<Func<EventData, Task>> _eventCallbacks = new List<Func<EventData, Task>>();

public void RegisterEventCallback(Func<EventData, Task> callback)
{
    _eventCallbacks.Add(callback);
}

// 在DataReceiver中调用回调
private async Task DataReceiver()
{
    while (true)
    {
        byte[] receivedData = await transport.ReceiveDataAsync();
        ParsedData parsed = Parse(receivedData);
        if (parsed is EventData eventData)
        {
            foreach (var callback in _eventCallbacks)
            {
                await callback(eventData);
            }
        }
    }
}

这种方式简单直接，适合小型场景，但灵活性不如前几种方案。

内容的提问来源于stack exchange，提问作者JimEvans