zoukankan      html  css  js  c++  java
  • 消息队列并发处理基类-简化版

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

        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

  • 相关阅读:
    delphi 属性 参数 新注释
    delphi query阻塞执行 长时间执行sql的解决办法
    FDLocalSQL
    C# Webservice
    vmware的centos 6虚拟机如何共享文件夹?
    tomcat如何配置启动时自动部署webapps下的war包
    tomcat如何配置context的docBase
    windows本地启动tomcat闪退
    jfinal 字节流存储文件以及解、压缩工具类
    java try catch 异常后还会继续执行吗
  • 原文地址:https://www.cnblogs.com/bighuiwolf/p/SimplifiedMessageQueueHandlerBase.html
Copyright © 2011-2022 走看看