Consumption BlockingCollection with multiple consumers

Question

I have the following program

 class Program
    {
        public static int TaskCount { get; set; }
        public static BlockingCollection<string> queue = new BlockingCollection<string>(new ConcurrentQueue<string>());
        static void Main(string[] args)
        {
            TaskCount = 3;
            Task.Factory.StartNew(() => Producer());

            for (int i = 0; i < TaskCount; i++)
                Task.Factory.StartNew(() => Consumer());
            Console.ReadKey();
        }

        private static void Producer()
        {
            using (StreamWriter sw = File.AppendText(@"C:\pcadder.txt"))
            {
                for (int i = 0; i < 15; i++)
                {
                    queue.Add("Item: " + (i+1).ToString());
                    var message = string.Format("{2}.Item added: Item {0} at {1}", (i+1).ToString(), DateTime.Now.ToString("yyyy/MM/dd hh:mm:ss.ffffff"),i+1);
                    Console.WriteLine(message);
                    sw.WriteLine(message);

                }
                queue.CompleteAdding();
            }
        }
        private static void Consumer()
        {
            int count = 1;
            foreach (var item in queue.GetConsumingEnumerable())
            {
                var message = string.Format("{3}.Item taken: {0} at {1} by thread {2}.", item, DateTime.Now.ToString("yyyy/MM/dd hh:mm:ss.ffffff"),
                        Thread.CurrentThread.ManagedThreadId,count);
                Console.WriteLine(message);

                using (StreamWriter sw = File.AppendText(@"C:\pctaker.txt"))
                    sw.WriteLine(message);
                count += 1;
            }
        }
    }

Output

1.Item added: Item 1 at 2017.07.06 09:58:49.784734
2.Item added: Item 2 at 2017.07.06 09:58:49.784734
3.Item added: Item 3 at 2017.07.06 09:58:49.784734
4.Item added: Item 4 at 2017.07.06 09:58:49.784734
5.Item added: Item 5 at 2017.07.06 09:58:49.784734
6.Item added: Item 6 at 2017.07.06 09:58:49.784734
7.Item added: Item 7 at 2017.07.06 09:58:49.784734
8.Item added: Item 8 at 2017.07.06 09:58:49.784734
9.Item added: Item 9 at 2017.07.06 09:58:49.784734
10.Item added: Item 10 at 2017.07.06 09:58:49.784734
11.Item added: Item 11 at 2017.07.06 09:58:49.784734
12.Item added: Item 12 at 2017.07.06 09:58:49.784734
13.Item added: Item 13 at 2017.07.06 09:58:49.784734
14.Item added: Item 14 at 2017.07.06 09:58:49.784734
15.Item added: Item 15 at 2017.07.06 09:58:49.784734

1.Item taken: Item: 3 at 2017.07.06 09:58:49.784734 by thread 7.
1.Item taken: Item: 2 at 2017.07.06 09:58:49.784734 by thread 4.
1.Item taken: Item: 1 at 2017.07.06 09:58:49.784734 by thread 5.
2.Item taken: Item: 5 at 2017.07.06 09:58:49.784734 by thread 4.
2.Item taken: Item: 4 at 2017.07.06 09:58:49.784734 by thread 7.
2.Item taken: Item: 6 at 2017.07.06 09:58:49.784734 by thread 5.
3.Item taken: Item: 7 at 2017.07.06 09:58:49.784734 by thread 4.
3.Item taken: Item: 8 at 2017.07.06 09:58:49.784734 by thread 7.
3.Item taken: Item: 9 at 2017.07.06 09:58:49.784734 by thread 5.
4.Item taken: Item: 11 at 2017.07.06 09:58:49.784734 by thread 7.
4.Item taken: Item: 12 at 2017.07.06 09:58:49.784734 by thread 5.
5.Item taken: Item: 13 at 2017.07.06 09:58:49.784734 by thread 7.
5.Item taken: Item: 14 at 2017.07.06 09:58:49.784734 by thread 5.
6.Item taken: Item: 15 at 2017.07.06 09:58:49.784734 by thread 7.

After almost every program start, I have one element that is not in the user logs. ( Item 10Missing here ). I could not understand why this is happening.

• How is this item not processed?
• When using multiple tasks as a consumer, are items in order (FIFO) spoiled processed? If I want to keep / forcefully process in FIFO order within the consumer method, should I avoid using multiple tasks? (Processing may include I / O, network operations)

Answer 1

using (StreamWriter sw = File.AppendText(@"C:\pctaker.txt"))
    sw.WriteLine(message);

. , . , , , . , . , , :

// create it outside `Consumer` and make synchronized
using (var taker = TextWriter.Synchronized(File.AppendText(@"pctaker.txt"))) {
    TaskCount = 3;
    Task.Factory.StartNew(() => Producer());
    //Producer();
    for (int i = 0; i < TaskCount; i++)
        // pass to consumer
        Task.Factory.StartNew(() => Consumer(taker));
    Console.ReadKey();
}

private static void Consumer(TextWriter writer)
{
    int count = 1;
    foreach (var item in queue.GetConsumingEnumerable())
    {
        var message = string.Format("{3}.Item taken: {0} at {1} by thread {2}.", item, DateTime.Now.ToString("yyyy/MM/dd hh:mm:ss.ffffff"),
                Thread.CurrentThread.ManagedThreadId, count);
        Console.WriteLine(message);                                
        writer.WriteLine(message);
        writer.Flush();
        count += 1;
    }
}

a lock .

- - FIFO, , , , , . A 1, B 2 . 100 1, 10 2. - 2 ( - ) 1.

Answer 2

, , , - , BlockingCollection.

, DataFlow library ( ).

. , await, DataFlow. - int, - int, .

, , , .

" X Y", , X , . , .

" X", , X , ( ).

, .

using System;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

namespace ConsoleApp1
{
    public class Program
    {
        static void Main()
        {
            var inQueue  = new TransformBlock<int, int>(item => process(item), processBlockOptions());
            var outQueue = new ActionBlock<int>(item => output(item), outputBlockOptions());

            inQueue.LinkTo(outQueue, new DataflowLinkOptions {PropagateCompletion = true});

            var task = queueData(inQueue);

            Console.WriteLine("Waiting for task to complete in thread " + Thread.CurrentThread.ManagedThreadId);
            task.Wait();
            Console.WriteLine("Completed.");
        }

        static async Task queueData(TransformBlock<int, int> executor)
        {
            await enqueue(executor);

            Console.WriteLine("Indicating that no more data will be queued.");
            executor.Complete(); // Indicate that no more items will be queued.

            Console.WriteLine("Waiting for queue to empty.");
            await executor.Completion; // Wait for executor queue to empty.
        }

        static async Task enqueue(TransformBlock<int, int> executor)
        {
            for (int i = 0; i < 100; ++i)
            {
                Console.WriteLine("Queuing data " + i);
                int v = i;
                await executor.SendAsync(v); // Queues a method that returns v.
            }
        }

        static int process(int value)  // Procss value by adding 1000 to it.
        {
            Console.WriteLine($"Thread {Thread.CurrentThread.ManagedThreadId} is processing item {value}");
            value += 1000;
            Thread.Sleep(150+nextRand());  // Simulate work.
            Console.WriteLine($"Returning {value} from thread {Thread.CurrentThread.ManagedThreadId}");

            return value;
        }

        static void output(int value)
        {
            Console.WriteLine($"Outputting {value}");
        }

        static ExecutionDataflowBlockOptions processBlockOptions()
        {
            return new ExecutionDataflowBlockOptions
            {
                MaxDegreeOfParallelism = 4,
                BoundedCapacity = 8
            };
        }

        static ExecutionDataflowBlockOptions outputBlockOptions()
        {
            return new ExecutionDataflowBlockOptions
            {
                MaxDegreeOfParallelism = 1,
                BoundedCapacity = 1
            };
        }

        static int nextRand()
        {
            lock (rngLock)
            {
                return rng.Next(250);
            }
        }

        static Random rng = new Random();
        static object rngLock = new object();
    }
}

MaxDegreeOfParallelism BoundedCapacity, processBlockOptions().

, MaxDegreeOfParallelism 8 BoundedCapacity = 16.

---

[EDIT] " ?" - , ( Evk)