如何用System.Threading.Channels实现单生产者多消费者接收相同消息？

问题根源 #

默认的Channel<T>是队列式分发机制，每条消息被任意一个消费者读取后就会从队列中移除，所以多个消费者会分摊接收消息，无法实现“所有消费者都收到全部消息”的需求。

实现方案 #

要实现消息广播，核心思路是给每个消费者分配独立的消息通道，生产者将同一条消息复制发送到所有通道，确保每个消费者都能拿到完整的消息流。以下提供两种可行方案：

方案1：独立Channel + 生产者手动广播 #

这种方式直接为每个消费者创建专属Channel，生产者遍历所有Channel发送消息，逻辑直观易懂：

public class DummyData
{
    public long Ticks { get; set; }
    public DateTime DateTime { get; set; }
    public decimal Price { get; set; }
}

// 通用消费者类，替代重复的ConsumerOne/ConsumerTwo
class Consumer
{
    private readonly Channel<DummyData> _tickQueue;
    private readonly CancellationTokenSource _cts;
    private readonly string _tag;

    public Consumer(Channel<DummyData> tickQueue, CancellationTokenSource cts, string tag)
    {
        _tickQueue = tickQueue;
        _cts = cts;
        _tag = tag;
    }

    public async Task StartConsuming()
    {
        await foreach (var message in _tickQueue.Reader.ReadAllAsync(_cts.Token))
        {
            // 替换为各自的策略逻辑
            Console.WriteLine($"from consumer {_tag} ==> {message.Price}");
        }
    }
}

class Producer
{
    private readonly List<Channel<DummyData>> _consumerChannels;
    private readonly CancellationTokenSource _cts;

    public Producer(List<Channel<DummyData>> consumerChannels, CancellationTokenSource cts)
    {
        _consumerChannels = consumerChannels;
        _cts = cts;
    }

    public async Task StartProducing()
    {
        Random r = new Random();
        for (int i = 0; i < 10; i++)
        {
            var data = new DummyData()
            {
                DateTime = DateTime.Now,
                Ticks = DateTime.Now.Ticks,
                Price = Convert.ToDecimal(r.NextDouble() * r.Next(100, 105))
            };

            // 向所有消费者通道发送同一条消息
            foreach (var channel in _consumerChannels)
            {
                await channel.Writer.WriteAsync(data, _cts.Token);
            }

            await Task.Delay(r.Next(50, 500));
        }

        // 发送完成后关闭所有通道的写入端
        foreach (var channel in _consumerChannels)
        {
            channel.Writer.Complete();
        }
    }
}

internal class MultipleConsumersEg
{
    private static readonly CancellationTokenSource _cts = new CancellationTokenSource();

    public static async Task Main(string[] args)
    {
        var consumerChannels = new List<Channel<DummyData>>();
        var consumers = new List<Consumer>();

        // 创建消费者1及其专属通道
        var channel1 = Channel.CreateUnbounded<DummyData>();
        consumerChannels.Add(channel1);
        consumers.Add(new Consumer(channel1, _cts, "ONE"));

        // 创建消费者2及其专属通道
        var channel2 = Channel.CreateUnbounded<DummyData>();
        consumerChannels.Add(channel2);
        consumers.Add(new Consumer(channel2, _cts, "TWO"));

        // 启动所有消费者
        foreach (var consumer in consumers)
        {
            _ = consumer.StartConsuming();
        }

        // 启动生产者
        var producer = new Producer(consumerChannels, _cts);
        await producer.StartProducing();

        Console.ReadLine();
        _cts.Cancel();
    }
}

方案2：封装广播器类（更优雅的动态订阅模式） #

如果消费者数量需要动态增减，可以封装一个广播器类，统一管理订阅和消息转发：

// 通用广播器，支持任意类型消息的订阅与广播
public class ChannelBroadcaster<T>
{
    private readonly List<Channel<T>> _subscribers = new List<Channel<T>>();
    private readonly object _lock = new object();

    // 消费者订阅，返回专属的消息读取器
    public ChannelReader<T> Subscribe()
    {
        var channel = Channel.CreateUnbounded<T>();
        lock (_lock)
        {
            _subscribers.Add(channel);
        }
        return channel.Reader;
    }

    // 广播消息到所有订阅者
    public async Task BroadcastAsync(T message, CancellationToken cancellationToken = default)
    {
        List<Channel<T>> subscribers;
        lock (_lock)
        {
            subscribers = new List<Channel<T>>(_subscribers);
        }

        foreach (var channel in subscribers)
        {
            await channel.Writer.WriteAsync(message, cancellationToken);
        }
    }

    // 停止广播并清理所有订阅通道
    public void Complete()
    {
        lock (_lock)
        {
            foreach (var channel in _subscribers)
            {
                channel.Writer.Complete();
            }
            _subscribers.Clear();
        }
    }
}

// 消费者类改为接收ChannelReader
class Consumer
{
    private readonly ChannelReader<DummyData> _reader;
    private readonly CancellationTokenSource _cts;
    private readonly string _tag;

    public Consumer(ChannelReader<DummyData> reader, CancellationTokenSource cts, string tag)
    {
        _reader = reader;
        _cts = cts;
        _tag = tag;
    }

    public async Task StartConsuming()
    {
        await foreach (var message in _reader.ReadAllAsync(_cts.Token))
        {
            Console.WriteLine($"from consumer {_tag} ==> {message.Price}");
        }
    }
}

class Producer
{
    private readonly ChannelBroadcaster<DummyData> _broadcaster;
    private readonly CancellationTokenSource _cts;

    public Producer(ChannelBroadcaster<DummyData> broadcaster, CancellationTokenSource cts)
    {
        _broadcaster = broadcaster;
        _cts = cts;
    }

    public async Task StartProducing()
    {
        Random r = new Random();
        for (int i = 0; i < 10; i++)
        {
            var data = new DummyData()
            {
                DateTime = DateTime.Now,
                Ticks = DateTime.Now.Ticks,
                Price = Convert.ToDecimal(r.NextDouble() * r.Next(100, 105))
            };

            // 只需调用一次广播，自动转发到所有订阅者
            await _broadcaster.BroadcastAsync(data, _cts.Token);

            await Task.Delay(r.Next(50, 500));
        }

        _broadcaster.Complete();
    }
}

internal class MultipleConsumersEg
{
    private static readonly CancellationTokenSource _cts = new CancellationTokenSource();

    public static async Task Main(string[] args)
    {
        var broadcaster = new ChannelBroadcaster<DummyData>();

        // 创建消费者并订阅广播器
        var consumer1 = new Consumer(broadcaster.Subscribe(), _cts, "ONE");
        var consumer2 = new Consumer(broadcaster.Subscribe(), _cts, "TWO");

        // 启动消费者
        _ = consumer1.StartConsuming();
        _ = consumer2.StartConsuming();

        // 启动生产者
        var producer = new Producer(broadcaster, _cts);
        await producer.StartProducing();

        Console.ReadLine();
        _cts.Cancel();
    }
}

注意事项 #

• 两种方案都能支持最多50个消费者的场景，Channel本身开销极低，只要消费者能及时处理消息，就不会出现性能问题。
• 如果消费者处理速度较慢，建议使用Channel.CreateBounded<T>设置通道容量上限，避免内存过度占用。
• 实际业务中可以根据消费者是否需要动态增减，选择对应的方案。

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