.NET管道应用——工作池

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

名词解释

工作池：一组等待任务分配的线程。一旦完成了所分配的任务，这些线程可继续等待任务的分配。

.NET管道：命名空间System.Threading.Channels中的Channel和Channel<T>对象，看nuget包，最低支持 .NETStandard 1.3

动机

最近在挑选.NET5项目动态编译运行字符串方案，最后选定了两个脚本语言库，想要测试一下高并发下的效率如何，就有了自己写个工作池来测试的念头。

Talk is cheap, show me the code.

/// <summary> /// 工作池 /// </summary> /// <param name="work">具体工作</param> /// <param name="workerCount">工人数</param> /// <param name="count">工作总件数</param> static async ValueTask WorkerPool<T>(Func<int, int, Task<T>> work, int workerCount, int count) { Channel<T> results = Channel.CreateUnbounded<T>(); //用于收集结果的不限容管道 Channel<int> workers = Channel.CreateBounded<int>(workerCount); //用于安排工作的限容管道,容量 = 工人数量(workerCount) var _ = ProcessResult(); //先开启收集结果的线程 var _ = 的作用在于 disable 4014警告 await ArrangeWork(); //等待安排工作并完成 async ValueTask ProcessResult() //处理结果 { for (var i = 0; i < count; i++) //读取结果次数 = 工作总件数 await results.Reader.ReadAsync(); //从结果管道读取结果，可以按照具体需求扩展 results.Writer.Complete(); // 先结束收集结果的管道 workers.Writer.Complete(); // 再结束安排工作的管道 } async ValueTask ArrangeWork() //安排工作 { var __ = Enumerable.Range(0, workerCount).Select(DoWork).ToArray(); //初始化工人 for (var i = 0; i < count; i++) await workers.Writer.WriteAsync(i); // await workers.Reader.Completion; //等待安排工作的管道完结。注：如果不等待，最后一批任务没有结束就返回了。 } async ValueTask DoWork(int workerId) //去工作 { await foreach (var order in workers.Reader.ReadAllAsync()) //应用异步Stream接取工作 { var rst = await work(workerId, order); //调用具体工作 await results.Writer.WriteAsync(rst); //写入工作成果到收集结果的管道 } } } static async ValueTask ChannelTest() { var sw = new Stopwatch(); sw.Start(); await WorkerPool(async (workerId, order) => { await Task.Delay(1000); WriteLine($"{sw.Elapsed:ss\\.fff}: {order}"); return true; }, 3, 9); //3个工人 9工作量 WriteLine($"{sw.Elapsed:ss\\.fff}: end"); }