using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace ConsoleApplication1
{
    class Program
    {
        static void Main(string[] args)
        {
            //var buffer1 = new BlockingCollection<int>(10); //Qtd maxima de 10 na pilha
            //var buffer2 = new BlockingCollection<int>(10);
            var buffer1 = new BlockingCollection<int>(); //Sem qtd maxima na pilha
            var buffer2 = new BlockingCollection<int>();

            var f = new TaskFactory(TaskCreationOptions.LongRunning,
                                    TaskContinuationOptions.None);

            //List<Task> t = new List<Task>();
            //t.Add(f.StartNew(() => CreateInitialRange(buffer1)));

            Task[] tasks = new Task[3];

            //Start the phases of the pipeline
            //Task stage1 = f.StartNew(() => CreateInitialRange(buffer1));
            //Task stage2 = f.StartNew(() => DoubleTheRange(buffer1, buffer2));
            //Task stage3 = f.StartNew(() => WriteResults(buffer2));

            tasks[0] = f.StartNew(() => CreateInitialRange(buffer1));
            tasks[1] = f.StartNew(() => DoubleTheRange(buffer1, buffer2));
            tasks[2] = f.StartNew(() => WriteResults(buffer2));

            //wait for the phases to complete
            //Task.WaitAll(stage1, stage2, stage3);
            Task.WaitAll(tasks);

            Console.WriteLine("Pressione uma tecla...");
            Console.ReadLine();
        }



        static void CreateInitialRange(BlockingCollection<int> output)
        {
            try
            {
                for (int i = 1; i <= 50; i++)
                {
                    Console.WriteLine("Inicio {0}, qtd {1}", i, output.Count());

                    //Console.WriteLine("CreateInitialRange {0}", i);
                    Thread.Sleep(200);
                    output.Add(i);
                }
            }
            finally
            {
                output.CompleteAdding();
            }
        }


        static void DoubleTheRange(BlockingCollection<int> input, BlockingCollection<int> output)
        {
            try
            {
                foreach (var number in input.GetConsumingEnumerable())
                {
                    Console.WriteLine("Meio {0}, qtd input {1}, qtd output {2}", number * number, input.Count(), output.Count());
                    Thread.Sleep(400);
                    output.Add(number * number);
                }
            }
            finally
            {
                output.CompleteAdding();
            }
        }


        static void WriteResults(BlockingCollection<int> input)
        {
            foreach (var squaredNumber in input.GetConsumingEnumerable())
            {
                //Console.WriteLine("Result is {0}", squaredNumber);
                Console.WriteLine("Fim {0}, qtd {1}", squaredNumber, input.Count());
                Thread.Sleep(800);
            }
        }

    }
}

Fonte: Link

0

0