消息队列处理基类-简化版

    public interface IMessageQueueHandler
    {
        void StartRead();
        int WorkerCount { get; }
    }

    public abstract class SimplifiedMessageQueueHandlerBase<T> : IMessageQueueHandler
    {
        public SimplifiedMessageQueueHandlerBase(string queueName)
        {
            if (!MessageQueue.Exists(queueName))
                throw new Exception();

            this._queueName = queueName;
        }

        public void StartRead()
        {
            this._queue = new MessageQueue(this._queueName) { Formatter = new XmlMessageFormatter(new Type[] { typeof(long) }) };
            this._queue.PeekCompleted += new PeekCompletedEventHandler(Produce);
            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); } }

        protected abstract string ProcessName { get; }

        protected abstract void MainProcess(T backThreadId);

        protected void LogInfo(string msg)
        {
            EntLibLogger.WriteLogFile(msg);
        }

        #region private
        private void Produce(object sender, PeekCompletedEventArgs e)
        {
            var message = this._queue.EndPeek(e.AsyncResult);
            T backThreadId = (T)message.Body;

            ThreadPool.QueueUserWorkItem(new WaitCallback(Consume), backThreadId);

            this._queue.Receive();
            this._queue.BeginPeek();
        }

        private void Consume(object stateInfo)
        {
            T messageItem = (T)stateInfo;
            this.LogInfo(string.Format("{0} - Received a message, MessageItem = {1}", this.ProcessName, messageItem));
            Interlocked.Increment(ref this._workerCount);

            try
            {
                this.LogInfo(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.LogInfo(string.Format("{0} - Over - {1}, WorkerCount = {2}", this.ProcessName, messageItem, this.WorkerCount));
            }
        }

        private void HandleException(Exception ex, T messageItem)
        {
            this.LogInfo(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;
        #endregion
    }