What thread-safe collection use to cache messages

Question

I'm working on project with following workflow :

• Background service consumme messages from Rabbitmq's queue
• Background service use background task queue like this and here to process task paralleling
• Each task execute queries to retrieve some datas and cache them in collection

If collection size is over 1000 objects, I would like to read collection and then clear it. Like each tasks are processing as parallel, I don't want that another thread add datas in collection until it was cleared.

There are blockingcollection or concurrentdictionary (thread-safe collection), but I don't know which mechanic to use ?

What's the best way to achieve this?

Answer 1

Check out concurrentQueue. It appears to be suitable for the tasks you have mentioned in your questions. Documentation here - https://learn.microsoft.com/en-us/dotnet/api/system.collections.concurrent.concurrentqueue-1?view=net-6.0

There are other concurrent collection types as well - https://learn.microsoft.com/en-us/dotnet/standard/collections/thread-safe/

Answer 2

​The collection that seems more suitable for your case is the Channel<T>. This is an asynchronous version of the BlockingCollection<T>, and internally it's based on the same storage (the ConcurrentQueue<T> collection)¹. The similarities are:

1. They both can be configured to be bounded or unbounded.
2. A consumer can take a message, even if none is currently available. In this case the Take/ReadAsync call will block either synchronously or asynchronously until a message can be consumed, or the collection completes, whatever comes first.
3. A producer can push a message, even if the collection is currently full. In this case the Add/WriteAsync call will block either synchronously or asynchronously until there is space available for the message, or the collection completes, whatever comes first.
4. A consumer can enumerate the collection in a consuming fashion, with a foreach/await foreach loop. Each message received in the loop is consumed by this loop, and will never be available to other consuming loops that might be active by other consumers in parallel.

Some features of the Channel<T> that the BlockingCollection<T> lacks:

1. A Channel<T> exposes two facades, a Writer and a Reader, that allow a better separation between the roles of the producer and the consumer. In practice this can be more of an annoyance than a useful feature IMHO, but nonetheless it's part of the experience of working with a channel.
2. A ChannelWriter<T> can be optionally completed with an error. This error is propagated to the consumers of the channel.
3. A ChannelReader<T> has a Completion property of type Task.
4. A bounded Channel<T> can be configured to be lossy, so that it drops old buffered messages automatically in order to make space for new incoming messages.

Some features of the BlockingCollection<T> that the Channel<T> lacks:

1. There is no direct support for timeout when writing/reading messages. This can be achieved indirectly (but precariously, see below) with timer-based CancellationTokenSources.
2. The contents of a channel cannot be enumerated in a non-consuming fashion.
3. Some auxiliary features like the BlockingCollection<T>.TakeFromAny method are not available.
4. A channel cannot be backed by other internal collections, other than the ConcurrentQueue<T>. So it can't have, for example, the behavior of a stack instead of a queue.

Other differences:

1. A BlockingCollection<T> is created using the constructor of the class. On the contrary a Channel<T> is created using the static factory methods of the Channel class.
2. The BlockingCollection<T> is not sealed, but it is practically not extendable (only the Dispose method is virtual and protected). It is possible though to inject custom functionality by passing in the constructor a custom IProducerConsumerCollection<T> implementation. On the contrary the Channel<T> is an abstract class, upon which derived classes can be built (if you are brave enough, and ready to handle the complexity of ValueTasks).

Caveat:

There is a nasty memory leak issue that is triggered when a channel is idle (empty with an idle producer, or full with an idle consumer), and at the other end a hyper-active consumer or producer attempts continuously to read/write messages with timer-based CancellationTokenSources. Each such canceled operation leaks about 800 bytes. The leak is resolved automatically when the first read/write operation completes successfully. This issue is known for more than two years, and Microsoft has not decided yet what to do with it.

¹ _{Actually the only channels that are backed by a ConcurrentQueue<T> are unbounded channels configured with SingleReader = false (the default). Currently there are three built-in implementations, two for unbounded channels (UnboundedChannel<T>, SingleConsumerUnboundedChannel<T>) and one for bounded channels (BoundedChannel<T>).}