惯性聚合 高效追踪和阅读你感兴趣的博客、新闻、科技资讯
阅读原文 在惯性聚合中打开

推荐订阅源

博客园 - 叶小钗
Microsoft Azure Blog
Microsoft Azure Blog
Stack Overflow Blog
Stack Overflow Blog
Jina AI
Jina AI
Vercel News
Vercel News
H
Help Net Security
Martin Fowler
Martin Fowler
美团技术团队
云风的 BLOG
云风的 BLOG
Y
Y Combinator Blog
阮一峰的网络日志
阮一峰的网络日志
MyScale Blog
MyScale Blog
钛媒体:引领未来商业与生活新知
钛媒体:引领未来商业与生活新知
博客园 - 三生石上(FineUI控件)
博客园 - 司徒正美
人人都是产品经理
人人都是产品经理
Engineering at Meta
Engineering at Meta
G
Google Developers Blog
Blog — PlanetScale
Blog — PlanetScale
MongoDB | Blog
MongoDB | Blog
宝玉的分享
宝玉的分享
小众软件
小众软件
T
Tailwind CSS Blog
WordPress大学
WordPress大学

博客园 - 灰灰狼

架构与设计概要 IoC概要 需求分析概要 接上文,支持并发数量的完美版本 2013年5.28~7.27 Microsoft FTE 微软面试总结 String Format for DateTime 多语言建议 multi-language 问题观 New life I would like About that task about wcf 基于证书的WCF安全开发详解 asp.net缓存(20100804完善版) - 灰灰狼 - 博客园 呼唤程序员精神——关于我今天发起的讨论的总结 asp.net mvc下实现窗口不关闭,就让Session不过期 正确的产品开发策略
消息队列并发处理基类-简化版
灰灰狼 · 2014-09-14 · via 博客园 - 灰灰狼

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

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

    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

posted on 2014-09-14 22:34  灰灰狼  阅读(2896)  评论()    收藏  举报