Q:

Multi Thread & Semaphore & Events

I am trying to execute some commands that come from RabbitMQ. Its about 5 msgs/sec. So as are too many msg, I have to send to a thread to execute, but I dont have so many threads, so I put a limit of 10. 
so the ideia was that the msgs would come to the worker, put in a queue and any of the 10 threads would peak and execute. All these using semaphore.
After some experiments, I don´t know why, but my thread only executes 3 or 4 items, after that it just stops with no error...
The problem I think is the logic when the event calls the method to execute, could not think in a better way...
Why just the first 4 msgs are processed??
What pattern or better way to do this?
Here are some parts of my code:
const int MaxThreads = 10;
private static Semaphore sem = new Semaphore(MaxThreads, MaxThreads);
private static Queue<BasicDeliverEventArgs> queue = new Queue<BasicDeliverEventArgs>();

static void Main(string[] args)
{
consumer.Received += (sender, ea) =>
               {
                var m = JsonConvert.DeserializeObject<Mail>(ea.Body.GetString());
                Console.WriteLine($"Sub-> {m.Subject}");
                queue.Enqueue(ea);
                RUN();
              };

            channel.BasicConsume(queueName, false, consumer);

            Console.Read();
}

private static void RUN()
{
            while (queue.Count > 0)
            {
                sem.WaitOne();
                var item = queue.Dequeue();
                ThreadPool.QueueUserWorkItem(sendmail, item);
            }
}

private static void sendmail(Object item)
{

//.....soem processing stuff....

//tell rabbitMq that everything was OK
channel.BasicAck(deliveryTag: x.DeliveryTag, multiple: true);

//release thread
sem.Release();

}

A:

I think that you could use a blocking collection here. It will simplify the code.
So your email sender would look something like that:
public class ParallelEmailSender : IDisposable
{
    private readonly BlockingCollection<string> blockingCollection;

    public ParallelEmailSender(int threadsCount)
    {
        blockingCollection = new BlockingCollection<string>(new ConcurrentQueue<string>());
        for (int i = 0; i < threadsCount; i++)
        {
            Task.Factory.StartNew(SendInternal);
        }
    }

    public void Send(string message)
    {
        blockingCollection.Add(message);
    }

    private void SendInternal()
    {
        foreach (string message in blockingCollection.GetConsumingEnumerable())
        {
            // send method
        }
    }

    public void Dispose()
    {
        blockingCollection.CompleteAdding();
    }
}

Of course you will need to add error catching logic and you could also improve the app shutting down process by using cancellation tokens.
I strongly suggest to read the great e-book about multithreading programming written by Joseph Albahari.

