ParallelProgramming-多消费者,多生产者同时运行并行

Posted C# java C++ 开源通信

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了ParallelProgramming-多消费者,多生产者同时运行并行相关的知识,希望对你有一定的参考价值。

在上一篇文章演示了并行的流水线操作(生产者和消费者并行同时执行),C#是通过BlockingCollection这个线程安全的对象作为Buffer,并且结合Task来实现的。但是上一篇文章有个缺陷,在整个流水线上,生产者和消费者是唯一的。本文将演示多个消费者多个生产者同时并行执行。

一、多消费者、多生产者示意图

技术分享

 与前一篇文章演示的流水线思想类似,不同之处就是本文的topic:消费者和生产者有多个,以buffer1为例,起生产者有两个,消费者有两个,现在有三个纬度的并行:

  1. Action1和Action2并行(消费者和生产者并行)
  2. 消费者并行(Action2.1和Action2.2并行)
  3. 生产者并行(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 代码解释

  1. Action11和Action12相对比较好理解。初始化数据到buffer1。
  2. Action2.1和Action2.2相对比较费解,他们同时接受buffer1作为输入,为什么最终的结果Buffer2没有产生重复? 最后由Action21,action22同时产生的buffer3为什么也没有重复?这就是GetConsumingEnumerable这个方法的功劳。这个方法会将buffer的数据分成多份给多个消费者,如果一个value已经被一个消费者获取,那么其他消费者将不会再拿到这个值。这就回答了为什么没有重复这个问题。
  3. 上面方法同时使用了多任务延续(ContinueWhenAll)对buffer的调用CompleteAdding方法:该方法非常重要,如果没有调用这个方法,程序会进入死锁,因为消费者(consumer)会处于一直的等待状态。

以上是关于ParallelProgramming-多消费者,多生产者同时运行并行的主要内容,如果未能解决你的问题,请参考以下文章

多生产者-多消费者问题

一、多线程下生产者消费者模式

多线程-线程间通信-多生产者多消费者示例

多线程_多生产者多消费者的实现问题

多线程

Java多线程系列--“基础篇”11之 生产消费者问题