C# Channel 简单实现消息队列的发布、订阅

十年河东，十年河西，莫欺少年穷

学无止境，精益求精

这里需要声明的是：

虽说 Channel 类似于 MQ ，但 Channel 并不支持跨项目使用，MQ 作为中间件，它里面的消息可以被多个项目共享，但  Channel 中的消息只能用于单体项目中共享。

介绍

首先，Channel本质上是.net中的一种新的集合类型，它与现有的Queue<T>类型非常相似，当然也有不同之处。

System.Threading.Channels 是.NET Core 3.0 后推出的新的集合类型, 具有异步API,高性能，线程安全等特点，它可以用来做消息队列，进行数据的生产和消费, 公开的 Writer 和 Reader api对应消息的生产者和消费者，也让Channel更加的简洁和易用，与Rabbit MQ 等其他队列不同的是，Channel 是进程内的队列。

为什么要使用 Channels

可以利用 Channels 来实现 生产者和消费者 之间的解耦，大体上有两个好处：

• 生产者 和 消费者 是相互独立的，两者可以并行执行。

• 如果生产者不给力，可以创建多个的生产者，如果消费者不给力，可以创建更多的消费者。

总的来说，在 生产者-消费者 模式下可以帮助我们提高应用程序的吞吐率。

创建Channel 

nuget 中搜索 Channel 并安装

channel 分为固定容量的，和无限容量的

无限容量的Channel创建

            //无限容量的
            var channel_1 = Channel.CreateUnbounded<msgDto>();

            var channel_2 = Channel.CreateUnbounded<string>(new UnboundedChannelOptions()
            {
                SingleWriter = false, //允许一次写入多个消息
                SingleReader = true //一次只能读取一条消息
            });

CreateUnbounded<T> 是指消息的类型，可以为简单的 int 、string 类型，也可以定义成一个类，上述的 msgDto 就是一个简答的类

    public class msgDto
    {
        public string msgid { get; set; }
        public string msg { get; set; }
    }

有限容量的Channel创建

            //限容Channel  
            var channel_3 = Channel.CreateBounded<string>(10000);
            var channel_4 = Channel.CreateBounded<msgDto>(new BoundedChannelOptions(100)
            {
                FullMode = BoundedChannelFullMode.Wait, //消息满了 等待消费 新的消息不插入
                SingleWriter = false,//允许一次写入多条数据
                SingleReader = false,//允许一次读取多条数据

            });

注意，BoundedChannelFullMode 共有四种

• Wait                   等待空间可用以便完成写入操作。

• DropWrite          删除要写入的项。

• DropNewest      删除并忽略通道中的最新项，以便为要写入的项留出空间。

• DropOldest        删除并忽略通道中的最旧项，以便为要写入的项留出空间。

生产/消费数据

        static async Task Main(string[] args)
        {
            //Channel 属于线程安全的集合
            var channel_1 = Channel.CreateUnbounded<string>(new UnboundedChannelOptions()
            {
                SingleWriter = false, //允许一次写入多条数据，下面的Parallel.For属于多线程写入
                SingleReader = false
            });
            //多线程写入 
            Parallel.For(0, 10, i =>
            {
                channel_1.Writer.WriteAsync ("hello_" + i);

            });

            //消费数据
            while (await channel_1.Reader.WaitToReadAsync())
            {
                if (channel_1.Reader.TryRead(out var message))
                {
                    Console.WriteLine(message);
                }
            }

            Console.Read();
        }

WaitToReadAsync是一个非阻塞等待，在有消息可读或Channel关闭时，才会唤醒并继续。

考虑到有多个消费者的情况，有可能别的线程已经进行了读取，这儿使用TryRead进行读取操作。

要注意：数据的同步工作是由Channel进行管理的。Channel会确保多个消费者不会读到相同的数据。Channel同时也管理数据的次序。

上述例子的输出结果：

 因为是异步多线程写入，因此，不保证顺序，但读取时，还是按照写入的顺序进行读取的。

参考：https://docs.microsoft.com/zh-cn/dotnet/api/system.threading.channels.boundedchannelfullmode?view=net-6.0

@天才卧龙的波尔卡 

只爱少妇