Occasionally I need to implement periodic checks in a loop in a background thread, a typical example being asynchronous processing using a message queue. While it’s not terribly complicated, I wonder what simpler, more elegant solutions exist.
I post my solution, and I’m eager to hear any constructive criticism!
The requirements in more details are these: I need:
- A loop that does periodic checks and actions if needed, and blocks in-between
- It runs in a background thread (since it does an active loop)
- It can be stopped properly (e.g. the loop exits and the thread freed)
- The blocking wait can be awaken if needed (e.g. new entry in queue, or exit)
- Thread safety
My solution:
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using System.Threading;
namespace BgLoopExample
{
/// <summary>
/// Perform asynchronous processing in a background thread.
/// Handle thread creation, exit, thread safety.
/// </summary>
public class BgLoopExample : IDisposable
{
/// <summary>
/// Queue for the incoming messages.
/// </summary>
private Queue<Message> myQueue;
/// <summary>
/// Task for background processing.
/// </summary>
private Task myBgLoopTask;
/// <summary>
/// Event for awakening the processing loop, used when new entry is added to the queue or exit requested.
/// </summary>
private AutoResetEvent myQueueEvent;
/// <summary>
/// Flag to signal stop for the parallel sender thread.
/// </summary>
private bool myStopTaskFlag;
/// <summary>
/// Flag to store if this class is disposing.
/// </summary>
private bool myDisposed;
/// <summary>
/// Class initializer, start the background thread.
/// </summary>
public void Init()
{
myQueue = new Queue<Message>();
myQueueEvent = new AutoResetEvent(false);
myStopTaskFlag = false;
myDisposed = false;
StartSenderTask();
}
/// <summary>
/// Adds new event to the Queue
/// </summary>
/// <param name="message">New event to add to the queue.</param>
public void AddMessage(Message message)
{
lock (myQueue)
{
if (myStopTaskFlag)
{
// don't add any more if stop has been requested
return;
}
myQueue?.Enqueue(message);
myQueueEvent?.Set();
}
}
/// <summary>
/// Starts the sender parallel task.
/// </summary>
private void StartSenderTask()
{
myStopTaskFlag = false;
myBgLoopTask = new Task(BgLoop);
myBgLoopTask.Start();
}
/// <summary>
/// Stops the sender parallel task, but send all data before exiting.
/// </summary>
private void StopSenderTask()
{
// set the stop flag, and signal change
myStopTaskFlag = true;
myQueueEvent.Set();
// wait till the loop exits
myBgLoopTask.Wait();
Console.WriteLine("Thread exited");
}
/// <summary>
/// Infinite loop for processing the incoming events from the Queue.
/// </summary>
private void BgLoop()
{
while (true)
{
// copy the message queue if it has entry (copy for thread safety)
List<Message> messageList = new List<Message>();
lock (myQueue)
{
if (myQueue.Count > 0)
{
while (myQueue.Count > 0)
{
messageList.Add(myQueue.Dequeue());
}
}
}
// Process the events
if (messageList.Count > 0)
{
ProcessMessages(messageList);
}
// Stop if need to stop and queue has been emptied
lock (myQueue)
{
if (myStopTaskFlag)
{
if (myQueue.Count == 0)
{
break;
}
// stop requested but not empty yet -- loop without waiting
continue;
}
}
// Block for a while. The exact timeout value does not really matter
myQueueEvent.WaitOne(10000);
}
Console.WriteLine("Thread exiting...");
}
/// <summary>
/// Send the next group of events using the data sender.
/// </summary>
/// <param name="messages">Event list to send.</param>
private void ProcessMessages(List<Message> messages)
{
foreach (Message m in messages)
{
try
{
Console.WriteLine($"{DateTime.Now}: Processing message {m} ...");
// ... do processing ...
}
catch (Exception ex)
{
// ...
}
}
}
public void Dispose()
{
Dispose(true);
GC.SuppressFinalize(this);
}
protected virtual void Dispose(bool disposing)
{
if (myDisposed)
{
return;
}
if (disposing)
{
StopSenderTask();
myQueueEvent.Dispose();
}
myDisposed = true;
}
}
}
Queue
useConcurrentQueue
and you can avoid your locks. See: msdn.microsoft.com/en-us/library/dd267265(v=vs.110).aspx Also this probably belongs better on codereview. \$\endgroup\$