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