多阅读器场景下System.Threading.Channels输出变慢问题排查

问题描述 #

测试System.Threading.Channels组件构建「发布-订阅」或「单生产者、多消费者」架构时，在WPF环境下发现：当存在多个通道阅读器（每个窗口对应一个订阅者/阅读器）时，数据输出会出现明显延迟。已在通道配置中设置SingleReader = false，但问题仍存在。

测试代码 #

public static class DataChannel
{
    private static readonly ConcurrentDictionary<string, Subscription<int>> Subscriptions = new();
    private static readonly ConcurrentDictionary<string, Task> Writers = new();

    public static ChannelReader<int> Subscribe(string key)
    {
        var subscription = Subscriptions.GetOrAdd(key, k => new Subscription<int>
        {
            Key = k,
            DataChannel = Channel.CreateUnbounded<int>(new UnboundedChannelOptions
            {
                SingleReader = false,
                SingleWriter = true
            })
        });

        Writers.GetOrAdd(key, _ => StartWriterTask(subscription.DataChannel.Writer));

        return subscription.DataChannel.Reader;
    }

    private static Task StartWriterTask(ChannelWriter<int> writer)
    {
        return Task.Factory.StartNew(async () =>
        {
            var i = 0;
            while (true)
            {
                await writer.WriteAsync(++i);
                await Task.Delay(500);
            }
        }, TaskCreationOptions.LongRunning);
    }
}

public partial class Subscription<T> : ObservableObject
{
    [ObservableProperty] private Channel<T> _dataChannel = null!;
    [ObservableProperty] private string _key = null!;
}


public partial class ConsumerWindowViewModel : ObservableObject
{
    [ObservableProperty] private int _dataValue;

    public ConsumerWindowViewModel()
    {
        _ = SubscribeForData();
    }


    private Task SubscribeForData()
    {
        Task.Factory.StartNew(async () =>
        {
            var reader = DataChannel.Subscribe("IntData");
            await foreach (var data in reader.ReadAllAsync())
            {
                DataValue = data;
            }
        }, TaskCreationOptions.LongRunning);

        return Task.CompletedTask;
    }
}

故障原因 #

你的实现根本不是发布-订阅模式，而是多个读者共享同一个通道的竞争消费模式。

System.Threading.Channels的核心规则是：每个消息只会被一个阅读器取走。当多个窗口的阅读器同时读取同一个通道时，它们会互相抢消息——第一个抢到的窗口会拿到当前数据，其他窗口只能等下一条消息，最终每个窗口只能收到间隔的数值，视觉上就表现为"延迟"。

设置SingleReader = false只是允许多个阅读器同时访问通道，但并没有改变"消息被消费即消失"的核心逻辑，所以问题无法解决。

修复方案：实现真正的发布-订阅 #

要让每个订阅者都收到完整的消息流，需要给每个订阅者分配独立的通道，生产者发送消息时广播到所有订阅者的通道。

修改后的核心代码示例：

public static class DataChannel
{
    // 维护每个主题对应的所有订阅者通道Writer
    private static readonly ConcurrentDictionary<string, List<ChannelWriter<int>>> _subscribers = new();
    private static readonly ConcurrentDictionary<string, Task> _writers = new();

    public static ChannelReader<int> Subscribe(string key)
    {
        // 每个订阅者创建独立的通道
        var channel = Channel.CreateUnbounded<int>(new UnboundedChannelOptions
        {
            SingleReader = true, // 单个订阅者用单读者模式更高效
            SingleWriter = true
        });

        // 将当前订阅者的Writer加入列表
        _subscribers.AddOrUpdate(key, 
            _ => new List<ChannelWriter<int>> { channel.Writer },
            (_, list) => { list.Add(channel.Writer); return list; });

        // 启动生产者任务（仅第一次订阅时启动）
        _writers.GetOrAdd(key, _ => StartWriterTask(key));

        return channel.Reader;
    }

    private static async Task StartWriterTask(string key)
    {
        var i = 0;
        while (true)
        {
            var data = ++i;
            // 把消息广播到所有订阅者的通道
            if (_subscribers.TryGetValue(key, out var writers))
            {
                // 遍历副本避免集合修改导致异常
                foreach (var writer in writers.ToList())
                {
                    if (!writer.TryWrite(data))
                    {
                        // 移除已关闭的订阅者（比如窗口已关闭）
                        writers.Remove(writer);
                    }
                }
            }
            await Task.Delay(500);
        }
    }
}

同时优化ViewModel的订阅逻辑，去掉不必要的Task.Factory.StartNew：

private async Task SubscribeForData()
{
    var reader = DataChannel.Subscribe("IntData");
    await foreach (var data in reader.ReadAllAsync())
    {
        DataValue = data;
    }
}

额外注意事项 #

• 当订阅者窗口关闭时，一定要调用ChannelWriter.Complete()，并从_subscribers中移除对应的Writer，避免内存泄漏
• 如果使用有界通道（Bounded Channel），要注意处理WriteAsync可能的阻塞情况，避免生产者被卡住

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