实现不考虑限制并发数的情况下对某队列的并发处理,欢迎批评指正:
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 }
觉得写的好的表扬一下啊:)
时间: 2024-10-10 21:39:55