Adding a listener event to a ConcurrentQueue or ConcurrentBag?
I have multiple tasks that grab messages from a queue in a 1:1 fashion. I want to add these messages from each thread into a ConcurrentBag and process them as they come in asynchronously. The purpose here is to get messages off the queues as quickly as possible so the queues do not fill up. I just need help with a listener that waits until messages are added to the ConcurrentBag then I need to remove the messages out of the Bag and process them
private static ConcurrentQueue<string> messageList = new ConcurrentQueue<string>();
private static readonly SemaphoreSlim semaphore = new SemaphoreSlim(50);
void Main (string[] args)
{
List<Task> taskList = new TaskList();
foreach(var job in JobList)
{
taskList.Add(Task.Run(() => ListenToQueue(job.QueueName));
}
Task.WaitAll(taskList.ToArray());
}
private async Task<string> ListenToQueue(string queueName)
{
var cancellationtoken = new CancellationTokenSource(TimeSpan.FromMinutes(60)).Token
//possibly 5000 messages can be on a single queue
while(!cancellationtoken.IsCancellationRequested)
{
var message = getMessageFromQueue(queueName);
messageList.Enqueue(message); //Add the message from each thread to a thread safe List
}
}
I Need a listener event here that each time something is added to the list then this event will fire. Also I need to remove the message from the list in a thread-safe way.
private void Listener()
{
var msg = string.Empty;
while (messageList.Count > 0)
{
messageList.TryDequeue(out msg)
await semaphore.WaitAsync();
Task.Run(() =>
{
try
{
if(!String.IsNullorEmpty(msg))
{
_ = ProcessMessage(msg); // I do not want to await anything but just fire and let it go
}
}
finally
{
sim.Release();
}
});
}
}
These days, I recommend going with an async-compatible solution like System.Threading.Channels
:
private static Channel<string> messageList = Channel.CreateUnbounded<string>();
private async Task<string> ListenToQueue(string queueName)
{
var cancellationtoken = new CancellationTokenSource(TimeSpan.FromMinutes(60)).Token;
try
{
var message = await getMessageFromQueue(queueName, cancellationtoken);
await messageList.Writer.WriteAsync(message, cancellationtoken);
}
catch (OperationCanceledException)
{
// ignored
}
}
private async Task Listener()
{
await foreach (var msg in messageList.Reader.ReadAllAsync())
{
if (!string.IsNullOrEmpty(msg))
_ = Task.Run(() => ProcessMessage(msg));
}
}
But if you want (or need) to stay in the blocking world, there's a solution there, too. ConcurrentBag<T>
and ConcurrentQueue<T>
are seldom used directly. Instead, it's more common to use BlockingCollection<T>
, which wraps a concurrent collection and provides a higher-level API, including GetConsumingEnumerable
:
private static BlockingCollection<string> messageList = new();
private async Task<string> ListenToQueue(string queueName)
{
var cancellationtoken = new CancellationTokenSource(TimeSpan.FromMinutes(60)).Token;
try
{
var message = await getMessageFromQueue(queueName, cancellationtoken);
messageList.Add(message, cancellationtoken);
}
catch (OperationCanceledException)
{
// ignored
}
}
private void Listener()
{
foreach (var msg in messageList.GetConsumingEnumerable())
{
if (!string.IsNullOrEmpty(msg))
_ = Task.Run(() => ProcessMessage(msg));
}
}