How to wrap ConcurrentDictionary in BlockingCollection?
blockingcollection, c#, concurrency, concurrentdictionary, producer-consumer
Solution
Maybe you need a concurrent dictionary of blockingCollection
ConcurrentDictionary<int, BlockingCollection<string>> mailBoxes = new ConcurrentDictionary<int, BlockingCollection<string>>();
int maxBoxes = 5;
CancellationTokenSource cancelationTokenSource = new CancellationTokenSource();
CancellationToken cancelationToken = cancelationTokenSource.Token;
Random rnd = new Random();
// Producer
Task.Factory.StartNew(() =>
{
while (true)
{
int index = rnd.Next(0, maxBoxes);
// put the letter in the mailbox 'index'
var box = mailBoxes.GetOrAdd(index, new BlockingCollection<string>());
box.Add("some message " + index, cancelationToken);
Console.WriteLine("Produced a letter to put in box " + index);
// Wait simulating a heavy production item.
Thread.Sleep(1000);
}
});
// Consumer 1
Task.Factory.StartNew(() =>
{
while (true)
{
int index = rnd.Next(0, maxBoxes);
// get the letter in the mailbox 'index'
var box = mailBoxes.GetOrAdd(index, new BlockingCollection<string>());
var message = box.Take(cancelationToken);
Console.WriteLine("Consumed 1: " + message);
// consume a item cost less than produce it:
Thread.Sleep(50);
}
});
// Consumer 2
Task.Factory.StartNew(() =>
{
while (true)
{
int index = rnd.Next(0, maxBoxes);
// get the letter in the mailbox 'index'
var box = mailBoxes.GetOrAdd(index, new BlockingCollection<string>());
var message = box.Take(cancelationToken);
Console.WriteLine("Consumed 2: " + message);
// consume a item cost less than produce it:
Thread.Sleep(50);
}
});
Console.ReadLine();
cancelationTokenSource.Cancel();
By this way, a consumer which is expecting something in the mailbox 5, will wait until the productor puts a letter in the mailbox 5.
Problem
I try to implement a `ConcurrentDictionary` by wrapping it in a `BlockingCollection` but did not seem to be successful. I understand that one variable declarations work with `BlockingCollection` such as `ConcurrentBag<T>`, `ConcurrentQueue<T>`, etc. So, to create a `ConcurrentBag` wrapped in a `BlockingCollection` I would declare and instantiate like this: ``` BlockingCollection<int> bag = new BlockingCollection<int>(new ConcurrentBag<int>()); ``` But how to do it for `ConcurrentDictionary`? I need the blocking functionality of the `BlockingCollection` on both the producer and consumer side.