public abstract class MessageQueueConcurrentHandlerBase<T> : IMessageQueueHandler    {        public MessageQueueConcurrentHandlerBase(string queueName, int maxConcurrency, Action<string> logDelegate)        {            if (!MessageQueue.Exists(queueName))                throw new Exception(string.Format("No such a queue: {0}", queueName));            if (maxConcurrency < 1)                throw new ArgumentOutOfRangeException("maxConcurrency");            this._queueName = queueName;            this._poolForConsumer = new Semaphore(0, maxConcurrency);            this._producerAutoResetEvent = new AutoResetEvent(false);            this._maxConcurrency = maxConcurrency;            this._logDelegate = logDelegate;        }        public void StartRead()        {            this._queue = new MessageQueue(this._queueName) { Formatter = new XmlMessageFormatter(new Type[] { typeof(long) }) };            this._queue.PeekCompleted += new PeekCompletedEventHandler(Produce);            this._producerAutoResetEvent.Set();            this._poolForConsumer.Release(this._maxConcurrency);            this._queue.BeginPeek();        }        public override string ToString()        {            return string.Format("{0}_{1}", this._queueName, this.ProcessName);        }        public int WorkerCount { get { return Thread.VolatileRead(ref this._workerCount); } }        public int MaxConcurrency { get { return _maxConcurrency; } }        protected abstract string ProcessName { get; }        protected abstract void MainProcess(T backThreadId);        protected Action<string> LogDelegate { get { return _logDelegate; } }        #region private        private void Produce(object sender, PeekCompletedEventArgs e)        {            this._producerAutoResetEvent.WaitOne();            var message = this._queue.EndPeek(e.AsyncResult);            ThreadPool.QueueUserWorkItem(new WaitCallback(this.Consume));            this._queue.BeginPeek();        }        private void Consume(object stateInfo)        {            this._poolForConsumer.WaitOne();            var message = this._queue.Receive();            this._producerAutoResetEvent.Set();            T messageItem = (T)message.Body;            this.LogDelegate(string.Format("{0} - Received a message, MessageItem = {1}", this.ProcessName, messageItem));            Interlocked.Increment(ref this._workerCount);            try            {                this.LogDelegate(string.Format("{0} - Running - {1}, WorkerCount = {2}", this.ProcessName, messageItem, this._workerCount));                MainProcess(messageItem);            }            catch (Exception ex)            {                this.HandleException(ex, messageItem);            }            finally            {                Interlocked.Decrement(ref this._workerCount);                this.LogDelegate(string.Format("{0} - Over - {1}, WorkerCount = {2}", this.ProcessName, messageItem, this._workerCount));            }            this._poolForConsumer.Release();        }        private void HandleException(Exception ex, T messageItem)        {            this.LogDelegate(string.Format("Exception in {0}:[Message]={1},[StackTrace]={2},[Type]={3},[_workerCount]={4},[backThreadId]={5}", this.ProcessName, ex.Message, ex.StackTrace, ex.GetType(), this.WorkerCount, messageItem));        }        private readonly string _queueName;        private MessageQueue _queue;        private int _workerCount;        private Semaphore _poolForConsumer;        private AutoResetEvent _producerAutoResetEvent;        private int _maxConcurrency;        private Action<string> _logDelegate;        #endregion    }