在上一篇文章演示了并行的流水线操作(生产者和消费者并行同时执行),C#是通过BlockingCollection这个线程安全的对象作为Buffer,并且结合Task来实现的。但是上一篇文章有个缺陷,在整个流水线上,生产者和消费者是唯一的。本文将演示多个消费者多个生产者同时并行执行。
一、多消费者、多生产者示意图
与前一篇文章演示的流水线思想类似,不同之处就是本文的topic:消费者和生产者有多个,以buffer1为例,起生产者有两个,消费者有两个,现在有三个纬度的并行:
- Action1和Action2并行(消费者和生产者并行)
- 消费者并行(Action2.1和Action2.2并行)
- 生产者并行(Action1.1和Action1.2并行)
二、实现
2.1 代码
class PiplelineDemo { PRivate int seed; public PiplelineDemo() { seed = 10; } public void Action11(BlockingCollection<string> output) { for (var i = 0; i < seed; i++) { output.Add(i.ToString());//initialize data to buffer1 } } public void Action12(BlockingCollection<string> output) { for (var i = 0; i < seed; i++) { output.Add(i.ToString());//initialize data to buffer1 } } public void Action21(BlockingCollection<string> input, BlockingCollection<string> output) { foreach (var item in input.GetConsumingEnumerable()) { var itemToInt = int.Parse(item); output.Add((itemToInt * itemToInt).ToString());// add new data to buffer2 } } public void Action22(BlockingCollection<string> input, BlockingCollection<string> output) { foreach (var item in input.GetConsumingEnumerable()) { var itemToInt = int.Parse(item); output.Add((itemToInt * itemToInt).ToString());// add new data to buffer2 } } public void Action31(BlockingCollection<string> input, BlockingCollection<string> output) { foreach (var item in input.GetConsumingEnumerable()) { output.Add((item));// add new data to buffer3 } } public void Action32(BlockingCollection<string> input, BlockingCollection<string> output) { foreach (var item in input.GetConsumingEnumerable()) { output.Add((item));// add new data to buffer3 } } public void Pipeline() { var buffer1 = new BlockingCollection<string>(seed * 2); var buffer2 = new BlockingCollection<string>(seed * 2); var buffer3 = new BlockingCollection<string>(seed * 2); var taskFactory = new TaskFactory(TaskCreationOptions.LongRunning, TaskContinuationOptions.None); var stage11 = taskFactory.StartNew(() => Action11(buffer1)); var stage12 = taskFactory.StartNew(() => Action12(buffer1)); Task.Factory.ContinueWhenAll(new Task[] { stage11, stage12 }, (tasks) => { buffer1.CompleteAdding(); }); var stage21 = taskFactory.StartNew(() => Action21(buffer1, buffer2)); var stage22 = taskFactory.StartNew(() => Action22(buffer1, buffer2)); Task.Factory.ContinueWhenAll(new Task[] { stage21, stage22 }, (tasks) => { buffer2.CompleteAdding(); }); var stage31 = taskFactory.StartNew(() => Action31(buffer2, buffer3)); var stage32 = taskFactory.StartNew(() => Action32(buffer2, buffer3)); Task.Factory.ContinueWhenAll(new Task[] { stage31, stage32 }, (tasks) => { buffer3.CompleteAdding(); }); Task.WaitAll(stage11, stage12, stage21, stage22, stage31, stage32); foreach (var item in buffer3.GetConsumingEnumerable())//print data in buffer3 { Console.WriteLine(item); } } }
2.2 运行结果
2.3 代码解释
- Action11和Action12相对比较好理解。初始化数据到buffer1。
- Action2.1和Action2.2相对比较费解,他们同时接受buffer1作为输入,为什么最终的结果Buffer2没有产生重复? 最后由Action21,action22同时产生的buffer3为什么也没有重复?这就是GetConsumingEnumerable这个方法的功劳。这个方法会将buffer的数据分成多份给多个消费者,如果一个value已经被一个消费者获取,那么其他消费者将不会再拿到这个值。这就回答了为什么没有重复这个问题。
- 上面方法同时使用了多任务延续(ContinueWhenAll)对buffer的调用CompleteAdding方法:该方法非常重要,如果没有调用这个方法,程序会进入死锁,因为消费者(consumer)会处于一直的等待状态。
时间: 2024-10-31 09:32:14