注册 登录  
 加关注
   显示下一条  |  关闭
温馨提示!由于新浪微博认证机制调整,您的新浪微博帐号绑定已过期,请重新绑定!立即重新绑定新浪微博》  |  关闭

天涯倦客的博客

祝福你朋友永远快乐!

 
 
 

日志

 
 

MSMQ稍复杂的包装类  

2012-10-11 12:10:45|  分类: asp.net |  标签: |举报 |字号 订阅

  下载LOFTER 我的照片书  |
继MSMQ简单包装类后,又把MSMQ再更新了一下。主要加入一些事件,有消息到达时,可用外部事件处理 (ProcessMessageHandler),以及消息格式不是预期格式时的事件(InvalidTypeHandler),设置接收超时时间 (Timeout);MSMQ里没有消息或接收超时的处理事件(NoMessageOrTimeoutHandler);可以设置接受到消息后是同步执行 (ThreadCount = 0时)还是异步执行(ThreadCount > 0时);并可以限制多线程执行时的数量(ThreadCount = n);

使用MSMQ的准备工作参见http://blog.csdn.net/iwteih/archive/2008/09/05/2884773.aspx

先上接口:
  1. namespace ×××.Msmq
  2. {
  3.     public interface IMsmq<T> : IDisposable
  4.     {
  5.         /// <summary>
  6.         /// Serializes and deserializes objects to or from the body of a message 
  7.         /// </summary>
  8.         MessageFormatter Formatter { getset; }

  9.         /// <summary>
  10.         /// The count of threads to process object in queue.
  11.         /// If omitted or 0, synchronous execution, otherwise asynchronous execution.
  12.         /// </summary>
  13.         int ThreadCount { getset; }

  14.         /// <summary>
  15.         /// The time to be spent to get a message from Msmq.
  16.         /// If omitted , the hook will be alive all the time
  17.         /// </summary>
  18.         TimeSpan Timeout { getset; }

  19.         /// <summary>
  20.         /// Push an object into MSMQ
  21.         /// </summary>
  22.         /// <param name="element">The object to be pushed into MSMQ</param>
  23.         void Push(T element);

  24.         /// <summary>
  25.         /// Pop the element in MSMQ
  26.         /// </summary>
  27.         /// <returns>object in msmq</returns>
  28.         T Pop();

  29.         /// <summary>
  30.         /// Start to listen Msmq
  31.         /// </summary>
  32.         void StartListener();

  33.         /// <summary>
  34.         /// Stop listening Msmq. Make sure that StopListen & StartListen are paired matched
  35.         /// or no StopListen.
  36.         /// </summary>
  37.         void StopListener();

  38.         /// <summary>
  39.         /// The real funtion to process messsage.
  40.         /// </summary>
  41.         event ProcessMessageHandler ProcessMessage;

  42.         /// <summary>
  43.         /// Triggered when no message in MSMQ or timeout
  44.         /// </summary>
  45.         event NoMessageOrTimeoutHandler ProcessNoMessageOrTimeout;

  46.         /// <summary>
  47.         /// If message type is expected, use this event for handling
  48.         /// </summary>
  49.         event InvalidTypeHandler ProcessInvalidType;
  50.     }
  51. }

实现如下:

  1. using System;
  2. using System.Collections.Generic;
  3. using System.Linq;
  4. using System.Text;
  5. using System.Messaging;
  6. using System.Threading;
  7. using System.Runtime.InteropServices;
  8. using log4net;

  9. namespace ×××.Msmq
  10. {
  11.     public enum MessageFormatter
  12.     {
  13.         XmlMessageFormatter,
  14.         BinaryMessageFormatter,
  15.     }


  16.     public class MessageArgs : EventArgs
  17.     {
  18.         /// <summary>
  19.         /// The object poped from queue
  20.         /// </summary>
  21.         public object ObjectToProcess { getset; }
  22.     }


  23.     public delegate void ProcessMessageHandler(object sender, MessageArgs args);

  24.     public delegate void NoMessageOrTimeoutHandler(object sender, EventArgs args);

  25.     public delegate void InvalidTypeHandler(object sender, EventArgs args);


  26.     internal class Msmq<T> : IMsmq<T>
  27.     {
  28.         private static readonly ILog logger = LogManager.GetLogger(
  29.                     System.Reflection.MethodBase.GetCurrentMethod().DeclaringType);

  30.         MessageQueue queue = null;

  31.         int upperLimit = 0;

  32.         int currentThreadCount = 0;


  33.         #region Constructor
  34.         /// <summary>
  35.         /// Initial a Msmq object
  36.         /// </summary>
  37.         /// <param name="qName">Queue name</param>
  38.         public Msmq(string qName)
  39.             : this(qName, 0)
  40.         {
  41.         }


  42.         /// <summary>
  43.         /// Initial a Msmq object.
  44.         /// </summary>
  45.         /// <param name="qName">Queue name</param>
  46.         /// <param name="limit">Upper limit in this msmq</param>
  47.         public Msmq(string qName, int limit)
  48.         {
  49.             if (!string.IsNullOrEmpty(qName) && !MessageQueue.Exists(qName))
  50.             {
  51.                 MessageQueue.Create(qName);
  52.             }
  53.             queue = new MessageQueue(qName);

  54.             upperLimit = limit;
  55.         }

  56.         #endregion

  57.         #region IMsmq<T> Members

  58.         IMessageFormatter msgFormatter = new XmlMessageFormatter(new Type[] { typeof(T) });
  59.         private MessageFormatter formatter;
  60.         /// <summary>
  61.         /// Serializes and deserializes objects to or from the body of a message 
  62.         /// </summary>
  63.         MessageFormatter IMsmq<T>.Formatter
  64.         {
  65.             get { return formatter; }
  66.             set
  67.             {
  68.                 formatter = value;
  69.                 if (formatter == MessageFormatter.BinaryMessageFormatter)
  70.                 {
  71.                     msgFormatter = new BinaryMessageFormatter();
  72.                 }
  73.                 else
  74.                 {
  75.                     msgFormatter = new XmlMessageFormatter(new Type[] { typeof(T) });
  76.                 }
  77.             }
  78.         }

  79.         /// <summary>
  80.         /// How many threads to process message. If omitted or 0, synchronous execution, otherwise asynchronous execution 
  81.         /// </summary>
  82.         private int threadCount = 0;
  83.         int IMsmq<T>.ThreadCount
  84.         {
  85.             get { return threadCount; }
  86.             set
  87.             {
  88.                 threadCount = value;
  89.                 if (threadCount < 0)
  90.                     threadCount = 0;
  91.             }
  92.         }

  93.         /// <summary>
  94.         /// The time to wait while trying to get an object from Msmq.
  95.         /// </summary>
  96.         private TimeSpan timeout = TimeSpan.MaxValue;
  97.         TimeSpan IMsmq<T>.Timeout
  98.         {
  99.             get { return timeout; }
  100.             set { timeout = value; }
  101.         }

  102.         private event ProcessMessageHandler internelProcessMessage;
  103.         /// <summary>
  104.         /// The real funtion to process messsage.
  105.         /// </summary>
  106.         //public event ProcessMessageHandler ProcessMessage;
  107.         event ProcessMessageHandler IMsmq<T>.ProcessMessage
  108.         {
  109.             add { internelProcessMessage += value; }
  110.             remove { internelProcessMessage -= value; }
  111.         }

  112.         private event NoMessageOrTimeoutHandler noMessageOrTimeoutHandler;
  113.         /// <summary>
  114.         /// Triggered when no message in MSMQ or timeout
  115.         /// </summary>
  116.         event NoMessageOrTimeoutHandler IMsmq<T>.ProcessNoMessageOrTimeout
  117.         {
  118.             add { noMessageOrTimeoutHandler += value; }
  119.             remove { noMessageOrTimeoutHandler -= value; }
  120.         }

  121.         private event InvalidTypeHandler invalidTypeHandler;
  122.         /// <summary>
  123.         /// If message type is expected, use this event for handling
  124.         /// </summary>
  125.         event InvalidTypeHandler IMsmq<T>.ProcessInvalidType
  126.         {
  127.             add { invalidTypeHandler += value; }
  128.             remove { invalidTypeHandler -= value; }
  129.         }

  130.         /// <summary>
  131.         /// Push an object into MSMQ
  132.         /// </summary>
  133.         /// <param name="element">The object to be pushed into MSMQ</param>
  134.         void IMsmq<T>.Push(T element)
  135.         {
  136.             Send(element);
  137.         }


  138.         void Send(object element)
  139.         {
  140.             using (System.Messaging.Message message = new System.Messaging.Message())
  141.             {
  142.                 message.Body = element;
  143.                 message.Formatter = msgFormatter;

  144.                 queue.Send(message);
  145.             }

  146.             //In original status, MSMQ is empty, error will throw when calling CurrentCount. Because queue is not open.
  147.             //so i allow insert action happen before calling CurrentCount.
  148.             //If reach the upper message limit number, sleep for one minute.
  149.             while (upperLimit != 0 && (CurrentMessageCount >= upperLimit))
  150.             {
  151.                 System.Threading.Thread.Sleep(60000);
  152.             }
  153.         }


  154.         /// <summary>
  155.         /// Pop the element in MSMQ
  156.         /// </summary>
  157.         /// <returns>object in msmq</returns>
  158.         T IMsmq<T>.Pop()
  159.         {
  160.             return Receive();
  161.         }

  162.         T Receive()
  163.         {
  164.             T element = default(T);
  165.             try
  166.             {
  167.                 using (Message message = queue.Receive(new TimeSpan(0, 0, 10)))
  168.                 {
  169.                     message.Formatter = msgFormatter;
  170.                     element = (T)message.Body;
  171.                 }
  172.             }
  173.             catch (MessageQueueException mqex)
  174.             {
  175.                 //Ingore the exception when queue is empty
  176.                 if (mqex.MessageQueueErrorCode != MessageQueueErrorCode.IOTimeout)
  177.                 {
  178.                     logger.Error(mqex);
  179.                 }
  180.             }
  181.             return element;
  182.         }

  183.         /// <summary>
  184.         /// Start to listen Msmq
  185.         /// </summary>
  186.         void IMsmq<T>.StartListener()
  187.         {
  188.             queue.ReceiveCompleted += new ReceiveCompletedEventHandler(queue_ReceiveCompleted);
  189.             BeginReceiveMessage();
  190.         }


  191.         /// <summary>
  192.         /// Stop listening Msmq. Make sure that StopListen & StartListen are paired matched
  193.         /// or no StopListen.
  194.         /// </summary>
  195.         void IMsmq<T>.StopListener()
  196.         {
  197.             queue.ReceiveCompleted -= new ReceiveCompletedEventHandler(queue_ReceiveCompleted);
  198.             currentThreadCount = 0;
  199.         }

  200.         #endregion


  201.         void BeginReceiveMessage()
  202.         {
  203.             queue.BeginReceive(timeout);
  204.         }

  205.         void queue_ReceiveCompleted(object sender, ReceiveCompletedEventArgs e)
  206.         {
  207.             Message msg = null;
  208.             MessageStatus status = MessageStatus.Unknown;
  209.             try
  210.             {
  211.                 msg = ((MessageQueue)sender).EndReceive(e.AsyncResult);
  212.                 status = MessageStatus.OK;
  213.             }
  214.             catch (MessageQueueException qexp)
  215.             {
  216.                 status = MessageStatus.QueueError;
  217.                 if (qexp.MessageQueueErrorCode == MessageQueueErrorCode.IOTimeout)
  218.                 {
  219.                     status = MessageStatus.IOTimeout;
  220.                 }
  221.                 else
  222.                 {
  223.                     logger.Error(qexp);
  224.                 }
  225.             }
  226.             catch (Exception exp)
  227.             {
  228.                 logger.Error(exp);
  229.             }

  230.             //Handle message in exception condition
  231.             if (msg == null)
  232.             {
  233.                 switch (status)
  234.                 {
  235.                     case MessageStatus.IOTimeout:
  236.                         if (noMessageOrTimeoutHandler != null)
  237.                         {
  238.                             noMessageOrTimeoutHandler(thisnull);
  239.                         }
  240.                         break;
  241.                     default:
  242.                         break;
  243.                 }
  244.                 BeginReceiveMessage();
  245.                 return;
  246.             }

  247.             msg.Formatter = msgFormatter;

  248.             if (msg.Body is T)
  249.             {
  250.                 T element = (T)msg.Body;

  251.                 if (internelProcessMessage != null)
  252.                 {
  253.                     //asynchronously execute
  254.                     if (threadCount > 0)
  255.                     {
  256.                         Action<T> callback = RunAsnyc;
  257.                         callback.BeginInvoke(element, AfterRunAsnyc, null);

  258.                         //control the multi-thread work flow to make sure only specified threads are working
  259.                         while (true)
  260.                         {
  261.                             if (currentThreadCount >= threadCount)
  262.                             {
  263.                                 Thread.Sleep(1000);
  264.                             }
  265.                             else
  266.                             {
  267.                                 Interlocked.Increment(ref currentThreadCount);
  268.                                 BeginReceiveMessage();
  269.                             }
  270.                         }
  271.                     }
  272.                     // synchronously execute
  273.                     else
  274.                     {
  275.                         internelProcessMessage(thisnew MessageArgs { ObjectToProcess = element });
  276.                         BeginReceiveMessage();
  277.                     }
  278.                 }
  279.             }
  280.             else
  281.             {
  282.                 status = Msmq<T>.MessageStatus.InvalidType;
  283.                 //If invalidTypeHandler defined, using it otherwise resend message into Msmq
  284.                 if (invalidTypeHandler != null)
  285.                 {
  286.                     invalidTypeHandler(msg, null);
  287.                 }
  288.                 else
  289.                 {
  290.                     Send(msg);
  291.                 }
  292.             }
  293.             msg.Dispose();
  294.         }

  295.         void AfterRunAsnyc(IAsyncResult itfAR)
  296.         {
  297.             Interlocked.Decrement(ref currentThreadCount);
  298.         }

  299.         private void RunAsnyc(T element)
  300.         {
  301.             foreach (var v in internelProcessMessage.GetInvocationList())
  302.             {
  303.                 ProcessMessageHandler pmh = (ProcessMessageHandler)v;
  304.                 pmh.Invoke(thisnew MessageArgs { ObjectToProcess = element });
  305.             }
  306.         }


  307.         /// <summary>
  308.         /// This works well for the situation that MSMQ and KEXQueue is on the same machine.
  309.         /// I am not sure it can work well if the two are separated.
  310.         /// </summary>
  311.         private int CurrentMessageCount
  312.         {
  313.             get
  314.             {
  315.                 //MSMQ.MSMQManagement msmq = new MSMQ.MSMQManagement();

  316.                 object server = null;
  317.                 object path = queue.Path;
  318.                 object format = null;

  319.                 //msmq.Init(ref server, ref path, ref format);

  320.                 //int count = msmq.MessageCount;

  321.                 //Marshal.ReleaseComObject(msmq);

  322.                 //return count;
  323.                 return 0;
  324.             }
  325.         }

  326.         enum MessageStatus
  327.         {
  328.             /// <summary>
  329.             /// Message received successfully
  330.             /// </summary>
  331.             OK,
  332.             /// <summary>
  333.             /// Queue is empty or occur when time out
  334.             /// </summary>
  335.             IOTimeout,
  336.             /// <summary>
  337.             /// Cannot convert the messsage to expected object type
  338.             /// </summary>
  339.             InvalidType,
  340.             /// <summary>
  341.             /// Exception occurs when receiving the message
  342.             /// </summary>
  343.             QueueError,
  344.             /// <summary>
  345.             /// Exception thrown not by Msmq
  346.             /// </summary>
  347.             Unknown
  348.         }

  349.         #region IDisposable Members

  350.         void IDisposable.Dispose()
  351.         {
  352.             if (queue != null)
  353.                 queue.Dispose();
  354.         }

  355.         #endregion
  356.     }
  357. }

 

  1. using System;
  2. using System.Collections.Generic;
  3. using System.Linq;
  4. using System.Text;

  5. namespace ×××.Msmq
  6. {    
  7.     public class MsmqFactory<T>
  8.     {
  9.         private static object objLock = new object();

  10.         private static Dictionary<string, IMsmq<T>> queuelist = new Dictionary<string, IMsmq<T>>();

  11.         /// <summary>
  12.         /// Create a Msmq instance.
  13.         /// </summary>
  14.         /// <param name="queueName">Queue name</param>
  15.         /// <returns>A Msmq instance</returns>
  16.         /// <remarks>If calling CreateMsmq twice. The two returned objects are separated instances.
  17.         /// For example, object one = CreateMsmq("A"); object two = CreateMsmq("A");
  18.         /// one and two are different objecct.
  19.         /// </remarks>
  20.         public static IMsmq<T> CreateMsmq(string queueName)
  21.         {
  22.             return new Msmq<T>(queueName);
  23.         }

  24.         /// <summary>
  25.         /// Create a Msmq instance with message count limitation.
  26.         /// </summary>
  27.         /// <param name="queueName">Queue name</param>
  28.         /// <param name="limit">Upper limit in this msmq</param>
  29.         /// <returns>A Msmq instance</returns>
  30.         /// <remarks>If calling CreateMsmq twice. The two returned objects are separated instances.
  31.         /// For example, object one = CreateMsmq("A"); object two = CreateMsmq("A");
  32.         /// one and two are different objecct.
  33.         /// </remarks>
  34.         public static IMsmq<T> CreateMsmq(string queueName, int limit)
  35.         {
  36.             return new Msmq<T>(queueName, limit);
  37.         }


  38.         /// <summary>
  39.         /// Initial a singleton Msmq object
  40.         /// </summary>
  41.         /// <param name="qName">Queue name</param>
  42.         /// <returns>A Msmq instance</returns>
  43.         /// <remarks>If calling CreateMsmq twice. The two returned objects are the same instances if their queue names are the same..
  44.         /// For example, object one = CreateMsmq("A"); object two = CreateMsmq("A");
  45.         /// one and two are the same object because they have the same queuename.
  46.         /// </remarks>
  47.         public static IMsmq<T> CreateSingletonMsmq(string queueName)
  48.         {
  49.             return CreateSingletonMsmq(queueName, 0);
  50.         }


  51.         /// <summary>
  52.         /// Initial a singleton Msmq object
  53.         /// </summary>
  54.         /// <param name="qName">Queue name</param>
  55.         /// <param name="limit">Upper limit in this msmq</param>
  56.         /// <returns>A Msmq instance</returns>
  57.         /// <remarks>If calling CreateMsmq twice. The two returned objects are the same instances if their queue names are the same.
  58.         /// For example, object one = CreateMsmq("A"); object two = CreateMsmq("A");
  59.         /// one and two are the same object because they have the same queuename.
  60.         /// </remarks>       
  61.         public static IMsmq<T> CreateSingletonMsmq(string queueName, int limit)
  62.         {
  63.             lock (objLock)
  64.             {
  65.                 if (!queuelist.ContainsKey(queueName))
  66.                 {
  67.                     Msmq<T> queue = new Msmq<T>(queueName);
  68.                     queuelist.Add(queueName, queue);
  69.                     return queue;
  70.                 }
  71.                 else
  72.                 {
  73.                     return queuelist[queueName];
  74.                 }
  75.             }
  76.         }


  77.         /// <summary>
  78.         /// Dispose a queue by given queuename.
  79.         /// </summary>
  80.         /// <param name="qName">The queue with the name to be disposed </param>
  81.         public static void DisposeQueue(string queueName)
  82.         {
  83.             lock (objLock)
  84.             {
  85.                 if (queuelist.ContainsKey(queueName))
  86.                 {
  87.                     queuelist[queueName].Dispose();
  88.                     queuelist.Remove(queueName);
  89.                 }
  90.             }
  91.         }

  92.         /// <summary>
  93.         /// Dispose all queues.
  94.         /// </summary>
  95.         public static void DisposeQueue()
  96.         {
  97.             lock (objLock)
  98.             {
  99.                 foreach (var queue in queuelist.Values)
  100.                 {
  101.                     queue.Dispose();
  102.                 }
  103.                 queuelist.Clear();
  104.             }
  105.         }

  106.     }
  107. }

调用如下:

  1. IMsmq<YourObj>  msmq = MsmqFactory<YourObj>.CreateMsmq("QueueName");
  2.             msmq.Formatter = MessageFormatter.XmlMessageFormatter;
  3.             msmq.Timeout = new TimeSpan(0, 0, 30);
  4.             msmq.ProcessMessage += new ProcessMessageHandler(ProcessMessage);
  5.             msmq.ProcessNoMessageOrTimeout += new NoMessageOrTimeoutHandler(ProcessNoMessageOrTimeout);

  6. msmq.StartListener();

MSMQ接收消息的方式有很多,比如BeginPeek或EndPeek,可以根据需要自行改动。

 

NOTE: 若将MSMQ定义成Transaction,则在多线程接收message时会出现消息丢失现象。
转载:http://blog.csdn.net/iwteih/article/details/3711079
  评论这张
 
阅读(644)| 评论(0)
推荐 转载

历史上的今天

在LOFTER的更多文章

评论

<#--最新日志,群博日志--> <#--推荐日志--> <#--引用记录--> <#--博主推荐--> <#--随机阅读--> <#--首页推荐--> <#--历史上的今天--> <#--被推荐日志--> <#--上一篇,下一篇--> <#-- 热度 --> <#-- 网易新闻广告 --> <#--右边模块结构--> <#--评论模块结构--> <#--引用模块结构--> <#--博主发起的投票-->
 
 
 
 
 
 
 
 
 
 
 
 
 
 

页脚

网易公司版权所有 ©1997-2017