一、流水线示意图

上图演示了流水线，action1接收input，然后产生结果保存在buffer1中，action2读取buffer1中由action1产生的数据，以此类推指导action4完成产生Output。

以上也是典型的生产者消费者模式。

上面的模式如果使用普通常规的串行执行是很简单的，按部就班按照流程图一步一步执行即可。如果为了提高效率，想使用并行执行，也就是说生产者和消费者同时并行执行，该怎么办么？

二、实现并行流水线

2.1 代码

class PiplelineDemo    {        private int seed;        public PiplelineDemo()        {            seed = 10;        }        public void Action1(BlockingCollection<string> output)        {            try            {                for (var i = 0; i < seed; i++)                {                    output.Add(i.ToString());//initialize data to buffer1                }            }            finally            {                output.CompleteAdding();            }        }        public void Action2(BlockingCollection<string> input, BlockingCollection<string> output)        {            try            {                foreach (var item in input.GetConsumingEnumerable())                {                    var itemToInt = int.Parse(item);                    output.Add((itemToInt * itemToInt).ToString());// add new data to buffer2                }            }            finally            {                output.CompleteAdding();            }        }        public void Action3(BlockingCollection<string> input, BlockingCollection<string> output)        {            try            {                foreach (var item in input.GetConsumingEnumerable())                {                    output.Add(item);//set data into buffer3                }            }            finally            {                output.CompleteAdding();            }        }        public void Pipeline()        {            var buffer1 = new BlockingCollection<string>(seed);            var buffer2 = new BlockingCollection<string>(seed);            var buffer3 = new BlockingCollection<string>(seed);            var taskFactory = new TaskFactory(TaskCreationOptions.LongRunning, TaskContinuationOptions.None);            var stage1 = taskFactory.StartNew(() => Action1(buffer1));            var stage2 = taskFactory.StartNew(() => Action2(buffer1, buffer2));            var stage3 = taskFactory.StartNew(() => Action3(buffer2, buffer3));            Task.WaitAll(stage1, stage2, stage3);            foreach(var item in buffer3.GetConsumingEnumerable())//print data in buffer3            {                Console.WriteLine(item);            }        }    }    class Program    {        static void Main(string[] args)        {            new PiplelineDemo().Pipeline();            Console.Read();        }    }

2.2 运行结果

预期打印出了0-9自我相乘的结果。

2.3 代码解释

代码本身的逻辑和本文开始的流程图是一一对应的。

BlockingCollection<T>是.Net里面的一个线程安全集合。实现了IProducerConsumerCollection<T>.

Add方法：将元素加入集合
CompleteAdding方法：告诉消费者，在当调用该方法之前的元素处理完之后就不要再等待处理了，可以结束处理了。这个非常重要，一定要执行，所以放在finally中（就算exception也要执行）
GetConsumingEnumberable，给消费者返回一个可以便利的集合

以上动图由“图斗罗”提供

本站仅提供存储服务，所有内容均由用户发布，如发现有害或侵权内容，请点击举报。