Asynchronous message execution is a common requirement of enterprise applications. Suppose that you have a web application which receives financial transaction requests, processes them (create accounting entries) and generates response.
Throughout the life cycle of a single transaction, many operations may need to be executed
without blocking current execution of request.For example, some kind of logging operations, notifying other integrated applications, etc.
This kind of non-blocking execution of operations can be fulfilled by way of message pools. In our type of message pool, there is one message pool in which the client code inserts its messages synchronously, but not executing messages.The message pool is listened by dedicated listener threads and responsibility of message execution belongs to them.
In this model, adding a message to the pool and taking a message from the pool may occur at the same time due to multithread model. For this reason, all the operations on pool must be thread safe.
For implementing this kind of message pool model, we can use ConcurrentQueue collection class of .NET Framework which is thread safe. Any kind of asynchronous execution request is enqueued by client code. Listener threads dequeue message one at a time, and invoke the "Execute" function of the message, which is contracted by an interface.
Here is the C# code:
Contract of the any message in the pool:
interface IThreadPoolMessage
{
void Execute();
}
--
Implementation of MessagePool class
Code:
class MessagePool : IDisposable
{
int listenerThreadCount;
ConcurrentQueue<IThreadPoolMessage> msgQueue;
bool continueProcessingMessages = true;
List<Thread> workerThreadList;
public MessagePool(int listenerThreadCount)
{
this.listenerThreadCount = listenerThreadCount;
Initialize();
}
void Initialize()
{
msgQueue = new ConcurrentQueue<IThreadPoolMessage>();
workerThreadList = new List<Thread>();
continueProcessingMessages = true;
for (int i = 0; i < listenerThreadCount; i++)
{
Thread th = new Thread(new ThreadStart(ListenPool));
th.Name = string.Format("NewThreadPool Thread_{0}", i + 1);
th.IsBackground = true;
workerThreadList.Add(th);
th.Start();
}
}
void ListenPool()
{
IThreadPoolMessage msg;
while (continueProcessingMessages)
{
if (msgQueue.TryDequeue(out msg))
{
try
{
msg.Execute();
}
catch (Exception ex)
{
}
}
}
}
public void AddMessage(IThreadPoolMessage msg)
{
this.msgQueue.Enqueue(msg);
}
public void StopAllThreads()
{
this.continueProcessingMessages = false;
while (!workerThreadList.TrueForAll(p => p.ThreadState == System.Threading.ThreadState.Stopped))
{
;
}
}
public bool HasAwatingMessage { get { return this.msgQueue.Any(); } }
public void Dispose()
{
StopAllThreads();
}
}
Here is a sample message, which is implemented for determining primality of an integer.
Code:
class CalculateIsPrimeMessage : IThreadPoolMessage
{
int val;
public CalculateIsPrimeMessage(int val)
{
this.val = val;
}
public void Execute()
{
bool isPrime = IsPrime(val);
WriteToDb(val, isPrime);
}
void WriteToDb(int val, bool isPrime)
{
using (SqlConnection conn = new SqlConnection("Server=.;Database=DENEME;Trusted_Connection=YES"))
{
string cmdStr = string.Format("INSERT INTO dbo.PRIMES SELECT {0}, {1}", val, isPrime ? 1 : 0);
using (SqlCommand cmd = new SqlCommand(cmdStr, conn))
{
conn.Open();
cmd.ExecuteNonQuery();
}
}
}
bool IsPrime(int val)
{
if (val % 2 == 0)
{
return val == 2;
}
else if (val % 3 == 0)
{
return val == 3;
}
else if (val % 5 == 0)
{
return val == 5;
}
for (int i = 7; i * i <= val; i += 2)
{
if (val % i == 0)
{
return false;
}
}
return true;
}
}
--
Here is the sample client code which used message pooling system:
Code:
static void Main(string[] args)
{
MessagePool msgPool = new MessagePool(10);
Stopwatch sw = Stopwatch.StartNew();
for (int i = 2; i < 10000; i++)
{
IThreadPoolMessage msg = new CalculateIsPrimeMessage(i);
msgPool.AddMessage(msg);
}
while (msgPool.HasAwatingMessage)
{
;
}
msgPool.Dispose();
sw.Stop();
Console.WriteLine("Elapsed Milliseconds: {0}", (int)sw.Elapsed.TotalMilliseconds);
}
--
