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

实现不考虑限制并发数的情况下对某队列的并发处理，欢迎批评指正：

    public interface IMessageQueueHandler
    {
        void StartRead();
        long 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 long WorkerCount { get { return Interlocked.Read(ref this._workerCount); } }

        protected abstract string ProcessName { get; }

        protected abstract void MainProcess(T messageItem);

        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 messageItem = (T)message.Body;

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

            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},[messageItem]={5}", this.ProcessName, ex.Message, ex.StackTrace, ex.GetType(), this.WorkerCount, messageItem));
        }

        private readonly string _queueName;
        private MessageQueue _queue;
        private long _workerCount;
        #endregion
    }

觉得写的好的表扬一下啊：），下一篇有完美版http://www.cnblogs.com/bighuiwolf/p/3972091.html