C#中通道(Channels)的应用之（生产者-消费者模式）

一.生产者-消费者模式概述

生产者-消费者模式是一种经典的设计模式，它将数据的生成（生产者）和处理（消费者）分离到不同的模块或线程中。这种模式的核心在于一个共享的缓冲区，生产者将数据放入缓冲区，而消费者从缓冲区中取出数据进行处理。这种模式有助于提高系统的响应性和吞吐量，因为它允许生产者和消费者并行工作，互不干扰。

二.Channels 概念

Channels提供了一种通信机制，允许生产者和消费者之间安全、可靠地交换信息，即使它们在不同的执行线程上运行。自.NET Core 3.0引入以来，System.Threading.Channels命名空间为我们处理生产者-消费者模式等复杂场景提供了强大的支持。Channels已经完全集成到.NET的异步模型中，支持async/await关键字，提供了一种异步的消息传递机制。通道本质上是一个线程安全的队列，支持在生产者和消费者之间安全、可靠地传递数据。通道有两种类型：有限容量的bound Channel和无限容量的unbound Channel。有限容量的通道在达到容量上限时会根据指定的策略处理新消息，而无限容量的通道则没有容量限制。

三.Channels 生产者-消费者模式实现

创建通道来作为生产者和消费者之间的共享缓冲区
  1. 无界通道
  • 无界容量的通道（即没有明确限制可以存储的项目数量的通道）,使用 Channel.CreateUnbounded<T>() 方法，如：
// 创建一个无界通道
var unboundedChannel = Channel.CreateUnbounded<string>();
  1. 有界通道
  • 创建有界通道则需要指定通道的容量上限，对于有限容量的通道，当通道满时，生产者可能需要等待或丢弃新数据。同样，当通道空时，消费者可能需要等待新数据的到来。通道提供了多种策略BoundedChannelFullMode 枚举处理方式：Wait：当通道已满时，写操作会等待直到队列中有空间来写入新的数据。这种情况下如果 TryWrite 操作会返回 false。DropOldest：如果通道已满，会删除最旧的数据（也就是最早进入通道但还未被读取的数据），以便给新的数据腾出空间。DropNewest：与 DropOldest 相反，会删除最新写入但还未被读取的数据来让新数据容纳进来。DropWrite：直接删除当前正在尝试写入的数据。
    使用 Channel.CreateBounded<T>(int capacity) 方法。例如：
// 创建一个 有界通道
var boundedChannel = Channel.CreateBounded<string>(100);
实现生产者
  • 生产者负责生成数据并将其写入通道。通常使用循环，在该循环中，生产者生成数据并使用WriteAsync方法将其写入通道。
async Task ProducerAsync(ChannelWriter<string> writer)
{for (int i = 0; i < 100; i++){await writer.WriteAsync(i.ToString());await Task.Delay(100); // 模拟数据生成的时间间隔}writer.Complete(); // 标记通道为完成写入，不再接受新数据
}
实现消费者
  • 消费者负责从通道中读取数据并进行处理。通常使用循环，在该循环中，消费者使用ReadAsync或ReadAllAsync方法从通道中读取数据，并对其进行处理。

async Task ConsumerAsync(ChannelReader<string> reader)
{while (await reader.WaitToReadAsync()){if (reader.TryRead(out var msgstring)){Console.WriteLine($"Consumed: {msgstring}");// 在这里处理数据}}
}

下面展示一个完整的生产者和消费者示例

  1. 启动 Program 类
// See https://aka.ms/new-console-template for more informationusing System.Threading.Channels;
using System.Threading.Tasks;
using TestChannels;Console.WriteLine("选择运行的模式?例如：1");
Console.WriteLine("1. 单生产单消费");
Console.WriteLine("2. 多生产单消费");
Console.WriteLine("3. 单生产多消费");
Console.WriteLine("4. 多生产多消费");
Console.WriteLine("请输入编号：");
var key = Console.ReadKey();switch (key.KeyChar)
{case '1':await SingleProducerSingleConsumer();break;case '2':await MultiProducerSingleConsumer();break;case '3':await SingleProduceMultipleConsumers();break;case '4':await MultiProducerMultipleConsumers();break;default:Console.WriteLine("请先选择运行模式!");break;
}// 单生产单消费
static async Task SingleProducerSingleConsumer()
{var channel = Channel.CreateUnbounded<string>();var producer1 = new Producer(channel.Writer, 1, 2000);var consumer1 = new Consumer(channel.Reader, 1, 1500);Task consumerTask1 = consumer1.ConsumerAsync(); // 开始消费Task producerTask1 = producer1.ProducerAsync(); // 开始生产await producerTask1.ContinueWith(_ => channel.Writer.Complete());await consumerTask1;
}// 多生产单消费
static async Task MultiProducerSingleConsumer()
{var channel = Channel.CreateUnbounded<string>();List<Task> producerTasks = new List<Task>();for (int i = 1; i <= 3; i++){producerTasks.Add(Task.Run(async () => {var producer = new Producer(channel.Writer, i, 2000);await producer.ProducerAsync();}));await Task.Delay(500); // 暂停500毫秒，启动另外一个生产}var consumer1 = new Consumer(channel.Reader, 1, 250);Task consumerTask1 = consumer1.ConsumerAsync(); // 开始消费await Task.WhenAll(producerTasks.ToArray()).ContinueWith(_ => channel.Writer.Complete());await consumerTask1;
}// 单生产多消费
static async Task SingleProduceMultipleConsumers()
{var channel = Channel.CreateUnbounded<string>();var producer1 = new Producer(channel.Writer, 1, 100);List<Task> consumerTasks = new List<Task>();for (int i = 1; i <= 3; i++){consumerTasks.Add(Task.Run(async () => {var consumer = new Consumer(channel.Reader, 1, 1500);await consumer.ConsumerAsync();}));}Task producerTask1 = producer1.ProducerAsync();await producerTask1.ContinueWith(_ => channel.Writer.Complete());await Task.WhenAll(consumerTasks.ToArray());
}// 多生产多消费
static async Task MultiProducerMultipleConsumers()
{var channel = Channel.CreateUnbounded<string>();List<Task> producerTasks = new List<Task>();for (int i = 1; i <=3; i++){Console.WriteLine("线程"+i.ToString());producerTasks.Add(Task.Run(async () => {var producer = new Producer(channel.Writer, i, 100);await producer.ProducerAsync();}));await Task.Delay(500); // 暂停500毫秒，启动另外一个生产}List<Task> consumerTasks = new List<Task>();for (int i = 1; i < 3; i++){consumerTasks.Add(Task.Run(async () => {var consumer = new Consumer(channel.Reader, 1, 1500);await consumer.ConsumerAsync();}));}await Task.WhenAll(producerTasks.ToArray()).ContinueWith(_ => channel.Writer.Complete());await Task.WhenAll(consumerTasks.ToArray());
}
  1. 生产者Producer 类
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Channels;
using System.Threading.Tasks;namespace TestChannels
{internal class Producer{private readonly ChannelWriter<string> _writer;private readonly int _identifier;private readonly int _delay;public Producer(ChannelWriter<string> writer, int identifier, int delay){_writer = writer;_identifier = identifier;_delay = delay;}public async Task ProducerAsync(){Console.WriteLine($"开始 ({_identifier}): 发布消息");for (var i = 0; i < 10; i++){await Task.Delay(_delay); // 停顿一下，方便观察数据var msg = $"P{_identifier} - {DateTime.Now:G}-{i}";Console.WriteLine($"发布 ({_identifier}): 消息成功 {msg}");await _writer.WriteAsync(msg);}Console.WriteLine($"发布 ({_identifier}): 完成");}}
}
  1. 消费者Consumer 类
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Channels;
using System.Threading.Tasks;namespace TestChannels
{/// <summary>/// 消费/// </summary>internal class Consumer{private readonly ChannelReader<string> _reader;private readonly int _identifier;private readonly int _delay;public Consumer(ChannelReader<string> reader, int identifier, int delay){_reader = reader;_identifier = identifier;_delay = delay;}public async Task ConsumerAsync(){Console.WriteLine($" 开始({_identifier}):消费 ");while (await _reader.WaitToReadAsync()){if (_reader.TryRead(out var timeString)){await Task.Delay(_delay); // 停顿一下，方便观察数据Console.WriteLine($"消费 ({_identifier}): 成功 {timeString}");}}Console.WriteLine($"消费 ({_identifier}): 完成");}}
}

运行

  • [ 参考] : https://learn.microsoft.com/en-us/dotnet/api/system.threading.channels?view=netcore-3.0

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/diannao/66868.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

计算机网络之---TCP连接管理

计算机网络之---TCP连接管理

TCP连接管理 TCP&#xff08;传输控制协议&#xff09;是面向连接的协议&#xff0c;在数据传输之前需要建立连接&#xff0c;在数据传输完成后需要断开连接。TCP连接的建立和断开都遵循特定的规则&#xff0c;分别称为三次握手&#xff08;Three-Way Handshake&#xff09;和四…
阅读更多...
【excel】VBA简介（Visual Basic for Applications）

【excel】VBA简介（Visual Basic for Applications）

文章目录 一、基本概念二、语法2.1 数据类型2.11 基本数据类型2.12 常量2.13 数组 2.2 控制语句2.21 条件语句2.22 循环语句2.23 错误处理&#xff1a;On Error2.24 逻辑运算 2.3 其它语句2.31 注释2.32 with语句 2.4 表达式2.41 常见表达式类型2.42 表达式的优先级 2.5 VBA 的…
阅读更多...
Git：Cherry-Pick 的使用场景及使用流程

Git：Cherry-Pick 的使用场景及使用流程

前面我们说了 Git合并、解决冲突、强行回退等解决方案 >> 点击查看 这里再说一下 Cherry-Pick功能&#xff0c;Cherry-Pick不是merge&#xff0c;只是把部分功能代码Cherry-Pick到远程的目标分支 git cherry-pick功能简介&#xff1a; git cherry-pick 是用来从一个分…
阅读更多...
机器学习模型调优指南

机器学习模型调优指南

机器学习模型调优指南 机器学习模型参数调优的作用在于优化模型的性能&#xff0c;使其能够在给定任务上更好地泛化和预测。通过合理调整模型的超参数&#xff0c;能够提高模型的准确性、降低过拟合或欠拟合的风险、加快训练过程等。具体来说&#xff0c;机器学习模型参数调优…
阅读更多...
mysql本地安装和pycharm链接数据库操作

mysql本地安装和pycharm链接数据库操作

MySQL本地安装和相关操作 Python相关&#xff1a;基础、函数、数据类型、面向、模块。 前端开发&#xff1a;HTML、CSS、JavaScript、jQuery。【静态页面】 Java前端&#xff1b; Python前端&#xff1b; Go前端 -> 【动态页面】直观&#xff1a; 静态&#xff0c;写死了…
阅读更多...
【Ubuntu与Linux操作系统：一、Ubuntu安装与基本使用】

【Ubuntu与Linux操作系统：一、Ubuntu安装与基本使用】

第1章 Ubuntu安装与基本使用 1.1 Linux与Ubuntu Linux是一种开源、类Unix操作系统内核&#xff0c;拥有高稳定性和强大的网络功能。由于其开源性和灵活性&#xff0c;Linux被广泛应用于服务器、嵌入式设备以及桌面环境中。 Ubuntu是基于Debian的一个流行Linux发行版&#xf…
阅读更多...
数据结构与算法之二叉树: LeetCode 654. 最大二叉树 (Ts版)

数据结构与算法之二叉树: LeetCode 654. 最大二叉树 (Ts版)

最大二叉树 https://leetcode.cn/problems/maximum-binary-tree/ 描述 给定一个不重复的整数数组 nums 。 最大二叉树 可以用下面的算法从 nums 递归地构建: 创建一个根节点&#xff0c;其值为 nums 中的最大值递归地在最大值 左边 的 子数组前缀上 构建左子树递归地在最大值…
阅读更多...
Pulsar客户端如何控制内存使用

Pulsar客户端如何控制内存使用

Pulsar客户端如何控制内存使用 一、使用场景 在实际应用中&#xff0c;Pulsar客户端的内存使用控制是一个重要的性能优化点。假设有一个搜索类业务需要记录用户搜索请求&#xff0c;以便后续分析搜索热点和优化搜索效果。以下是一个简化的代码示例&#xff1a; PulsarClient…
阅读更多...
【Unity高级】一文了解Unity 中的条件编译（附所有指令）

【Unity高级】一文了解Unity 中的条件编译（附所有指令）

一、Unity中的条件编译 Unity 对 C# 语言的支持包括使用指令&#xff0c;这些指令允许您根据是否定义了某些脚本符号&#xff0c;选择性地包含或排除代码的编译。有关这些指令在 C# 中如何工作的更多信息&#xff0c;请参阅微软关于 C# 预处理器指令 的文档。 &#xff08;一…
阅读更多...
苍穹外卖08——（涉及接收日期格式数据、ApachePOI导出报表、sql获取top10菜品数据）

苍穹外卖08——（涉及接收日期格式数据、ApachePOI导出报表、sql获取top10菜品数据）

营业额统计 service层 在需要处理空值、与数据库交互或使用集合时&#xff0c;Integer 、Double是更好的选择。 // 导入string工具类 import org.apache.commons.lang.StringUtils; Service // 标记该类为Spring的服务组件 Slf4j // 引入日志功能 public class Repor…
阅读更多...
微信小程序订阅消息提醒-云函数

微信小程序订阅消息提醒-云函数

微信小程序消息订阅分2种&#xff1a; 1.一次性订阅&#xff1a;用户订阅一次就可以推送一次&#xff0c;如果需要多次提醒需要多次订阅。 2.长期订阅&#xff1a;只有公共服务领域&#xff0c;如政务、医疗、交通、金融和教育等。‌在用户订阅后&#xff0c;在很长一段时间内…
阅读更多...
青少年编程与数学 02-006 前端开发框架VUE 22课题、状态管理

青少年编程与数学 02-006 前端开发框架VUE 22课题、状态管理

青少年编程与数学 02-006 前端开发框架VUE 22课题、状态管理 一、状态管理二、Vuex1. 安装Vuex2. 创建Vuex Store3. 在Vue应用中使用Store4. 在组件中使用状态5. 模块化Store 三、Vuex应用示例1. 创建项目2. 安装Vuex3. 设置Vuex Store4. 在主项目中使用Store5. 创建组件6. 更新…
阅读更多...
代码随想录算法训练营第 4 天（链表 2）| 24. 两两交换链表中的节点19.删除链表的倒数第N个节点 -

代码随想录算法训练营第 4 天（链表 2）| 24. 两两交换链表中的节点19.删除链表的倒数第N个节点 -

一、24. 两两交换链表中的节点 题目&#xff1a;24. 两两交换链表中的节点 - 力扣&#xff08;LeetCode&#xff09; 视频&#xff1a;帮你把链表细节学清楚&#xff01; | LeetCode&#xff1a;24. 两两交换链表中的节点_哔哩哔哩_bilibili 讲解&#xff1a;代码随想录 dummy-…
阅读更多...
pycharm-pyspark 环境安装

pycharm-pyspark 环境安装

1、环境准备&#xff1a;java、scala、pyspark、python-anaconda、pycharm vi ~/.bash_profile export SCALA_HOME/Users/xunyongsun/Documents/scala-2.13.0 export PATH P A T H : PATH: PATH:SCALA_HOME/bin export SPARK_HOME/Users/xunyongsun/Documents/spark-3.5.4-bin…
阅读更多...
数据结构与算法之链表: LeetCode 146. LRU 缓存 (Ts版)

数据结构与算法之链表: LeetCode 146. LRU 缓存 (Ts版)

LRU 缓存 https://leetcode.cn/problems/lru-cache/description/ 描述 请你设计并实现一个满足 LRU (最近最少使用) 缓存 约束的数据结构 实现 LRUCache 类&#xff1a; LRUCache(int capacity) 以 正整数 作为容量 capacity 初始化 LRU 缓存int get(int key) 如果关键字 ke…
阅读更多...
Three.js 渲染技术：打造逼真3D体验的幕后功臣

Three.js 渲染技术：打造逼真3D体验的幕后功臣

文章目录 前言一、着色器&#xff08;Shaders&#xff09;二、后处理&#xff08;Post-processing&#xff09;三、抗锯齿&#xff08;Anti-aliasing&#xff09;四、实时渲染与离线渲染五、光照模型与材质优化六、环境映射&#xff08;Environment Mapping&#xff09;七、纹理…
阅读更多...
EFCore HasDefaultValueSql (续1 ValueGeneratedOnAdd)

EFCore HasDefaultValueSql (续1 ValueGeneratedOnAdd)

前情&#xff1a;EFCore HasDefaultValueSql 小伙伴在使用 HasDefaultValueSql 时&#xff0c;对相关的 ValueGeneratedOnAdd 也有了疑问&#xff1a; ValueGeneratedOnAdd 和 HasDefaultValueSql 是 Entity Framework Core 中用于管理字段默认值的两种不同配置方式&#xff0…
阅读更多...
通过Apache、Nginx限制直接访问public下的静态文件

通过Apache、Nginx限制直接访问public下的静态文件

一、Apache 在public目录下的.htaccess文件中添加如下规则&#xff0c;来拒绝除了指定文件类型之外的所有请求 <FilesMatch "\.(?!(jpg|jpeg|png|gif|css|js|ico)$)[^.]$">Order Allow,DenyDeny from all </FilesMatch> 上述配置表示仅允许访问.jpg …
阅读更多...
远程和本地文件的互相同步

远程和本地文件的互相同步

文章目录 1、rsync实现类似git push pull功能1. 基础概念2. 示例操作3. 定制化和进阶用法4. 定时同步&#xff08;类似自动化&#xff09; 2 命令简化1. 动态传参的脚本2. Shell 函数支持动态路径3. 结合环境变量和参数&#xff08;更简洁&#xff09;4. Makefile 支持动态路径…
阅读更多...
AIOps 平台

AIOps 平台

AIOps&#xff08;Artificial Intelligence for IT Operations&#xff09;平台是一种结合人工智能&#xff08;AI&#xff09;技术和IT运营管理的解决方案&#xff0c;旨在通过自动化、智能化的手段优化企业IT系统的运行与管理。以下是AIOps平台的核心功能、优势以及常见的技术…
阅读更多...
最新文章

Copyright @ 2022~2023