当前位置 : 主页 > 编程语言 > 其它开发 >

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

来源:互联网 收集:自由互联 发布时间:2022-07-17
十年河东,十年河西,莫欺少年穷 学无止境,精益求精 这里需要声明的是: 虽说Channel 类似于 MQ ,但Channel 并不支持跨项目使用,MQ 作为中间件,它里面的消息可以被多个项目共享,

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

学无止境,精益求精

这里需要声明的是:

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

介绍

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

System.Threading.Channels 是.NET Core 3.0 后推出的新的集合类型, 具有异步API,高性能,线程安全等特点,它可以用来做消息队列,进行数据的生产和消费, 公开的 WriterReader 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

@天才卧龙的波尔卡 

只爱少妇
网友评论