Parallel Programming-实现并行操作的流水线(生产者、消费者)_.NET_编程开发_程序员俱乐部

中国优秀的程序员网站程序员频道CXYCLUB技术地图
热搜:
更多>>
 
您所在的位置: 程序员俱乐部 > 编程开发 > .NET > Parallel Programming-实现并行操作的流水线(生产者、消费者)

Parallel Programming-实现并行操作的流水线(生产者、消费者)

 2017/11/14 18:03:43  浅苍蓝  程序员俱乐部  我要评论(0)
  • 摘要:本文介绍如何使用C#实现并行执行的流水线(生产者消费者):1.流水线示意图2.实现并行流水线一、流水线示意图上图演示了流水线,action1接收input,然后产生结果保存在buffer1中,action2读取buffer1中由action1产生的数据,以此类推指导action4完成产生Output。以上也是典型的生产者消费者模式。上面的模式如果使用普通常规的串行执行是很简单的,按部就班按照流程图一步一步执行即可。如果为了提高效率,想使用并行执行,也就是说生产者和消费者同时并行执行,该怎么办么
  • 标签:实现 all 操作 消费者

本文介绍如何使用C#实现并行执行的流水线(生产者消费者):

1.流水线示意图

2.实现并行流水线

一、流水线示意图

 

上图演示了流水线,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>.

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

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

 

发表评论
用户名: 匿名