ASP.NET中进行消息处理(MSMQ) 三

分手后的思念是犯贱 2021-12-10 07:19 204阅读 0赞

在本文的前两篇文章里对MSMQ的相关知识点进行了介绍，很多阅读过这前两篇文章的朋友都曾问到过这样一些问题：  
  1、如何把MSMQ应用到实际的项目中去呢？  
  2、可不可以介绍一个实际的应用实例？  
  3、......

在前两篇文章里，关于MSMQ常用的技术点基本介绍完毕了，本文主要以MS开源项目PetShop中的MSMQ应用作为案例来介绍MSMQ在实际项目中的应用。在PetShop里，由于系统使用了多线程的专用应用程序来监控消息队列，在进入PetShop应用分析前，我们先来了解下关于多线程和MSMQ的相关知识点。  
  
**一、多线程和MSMQ**  
      现在有这样一个需求，指定的消息队列里不管有无消息数据，我们通过一个多线程来监控这个队列，一但队列里的数据发生变化就做出相应的处理，比如把消息读取出来。根据这个需求，我们来做个示例，用一多线程把队列监控起来，如果队列里有消息数据，就把消息读取出来，如果没有则一直监视队列，当队列数据发生改变(有新的消息加入)的时候就作出处理（读取消息）。  
  
      首先定义一个线程数组用于存储线程数；

1 ![None.gif][] static   private   int  ThreadNumber  =   5 ;   // 5个线程序   
 2 ![None.gif][] static   private  Thread\[\] ThreadArray  =   new  Thread\[ThreadNumber\];

我们把需要启动的线程装载入ThreadArray数组，通过一个遍历数组把所以的线程启动，实际这里只有5个线程。

1 ![None.gif][] private   void  button1\_Click( object  sender, EventArgs e)  
  2 ![ExpandedBlockStart.gif][]![ContractedBlock.gif][] ![dot.gif][] \{  
 3![InBlock.gif][]    StartThreads();  
 4![ExpandedBlockEnd.gif][]\}   
  5 ![None.gif][]  
  6 ![None.gif][] private   void  StartThreads()  
  7 ![ExpandedBlockStart.gif][]![ContractedBlock.gif][] ![dot.gif][] \{  
 8![InBlock.gif][]    int counter; //线程计数  
 9![InBlock.gif][]    for (counter = 0; counter < ThreadNumber; counter\++)  
10![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]    ![dot.gif][]\{  
11![InBlock.gif][]        ThreadArray\[counter\] = new Thread(new ThreadStart(MSMQListen));  
12![InBlock.gif][]        ThreadArray\[counter\].Start();  
13![InBlock.gif][]        this.richTextBox2.Text \+= (counter \+ 1).ToString() \+ "号线程开始！";  
14![ExpandedSubBlockEnd.gif][]    \}  
15![ExpandedBlockEnd.gif][]\}   
 16 ![None.gif][]  
 17 ![None.gif][] private   void  MSMQListen()  
 18 ![ExpandedBlockStart.gif][]![ContractedBlock.gif][] ![dot.gif][] \{  
19![InBlock.gif][]    while (true)  
20![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]    ![dot.gif][]\{  
21![InBlock.gif][]        //取出队列里的消息  
22![InBlock.gif][]        MessageBox.Show(MsgQueue.ReceiveMessage());  
23![ExpandedSubBlockEnd.gif][]    \}  
24![ExpandedBlockEnd.gif][]\}

如果上面这段代码阅读起存在问题，建议先去了解下多线程的相关知识点。在StartThreads方法里启动数组里存储的所以线程，并委托给MSMQListen方法进行处理，MSMQListen方法完成的就是读取队列里的消息，这里我使用了在第二篇文章里所使用的MsgQueue类和Book类，详细请阅读第二篇文章[ASP.NET中进行消息处理(MSMQ) 二][ASP.NET_MSMQ_] 。  
                              ![msmq9.jpg][]  
      启动了5个线程，用来监视指定的消息队列，如上图。那好，我们现在就来测试一下，通过给队列里发送消息，看线程是否会有响应。从上面启动线程的代码上可以很清晰的看出，只要队列里有消息在多线程的监视下，线程就会把队列里的消息读取出来。

1 ![None.gif][] private   void  button3\_Click( object  sender, EventArgs e)  
  2 ![ExpandedBlockStart.gif][]![ContractedBlock.gif][] ![dot.gif][] \{  
 3![InBlock.gif][]    Book book = new Book();  
 4![InBlock.gif][]    book.BookId = 1;  
 5![InBlock.gif][]    book.BookName = "asp.net";  
 6![InBlock.gif][]    book.BookAuthor = "abcd";  
 7![InBlock.gif][]    book.BookPrice = 50.80;  
 8![InBlock.gif][]  
 9![InBlock.gif][]    MsgQueue.SendMessage(book);  
10![ExpandedBlockEnd.gif][]\}

那么这里的测试，向队列里发送了一Book对象消息，根据上面分析，则多线程便会同时把此条消息读取出来：  
                            ![msmq10.jpg][]  
     呵呵，测试结果出来，看来此测试达到了我们之前所提出的需求。![tongue_smile.gif][]！OK，关于MSMQ和多线程就简单介绍于此。  
  
**二、MSMQ在开塬项目PetShop中的应用分析。**  
      在PetShop 4.0中，利用消息队列临时存放要插入的数据，来避免因为频繁访问数据库的操作。而队列中的消息，则等待系统的专用的应用程序来处理，最后将数据插入到数据库中。  
      PetShop 4.0中的消息处理，主要分为下面几大部分：订单策略接口IOrderStategy、消息接口IMessageing、消息工厂MessageFactory、MSMQ实现MSMQMessaging、后台处理应用程序OrderProessor。如下图：  
                                            ![petshop0.jpg][]  
     
1、订单策略接口IOrderStategy  
     PetShop 4.0的体系结构是非常庞大，在订单处理上有两种处理策略，这里也是策略模式的一个应用，IOrderStrategy接口作为订单策略的高层抽象，实现不同订单处理的具体策略去实现它，UML如下：   
                             ![petshop1.jpg][]

示意性代码：

1 ![None.gif][] namespace  PetShop.IBLLStrategy  
  2 ![ExpandedBlockStart.gif][]![ContractedBlock.gif][] ![dot.gif][] \{  
 3![InBlock.gif][]    public interface IOrderStrategy  
 4![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]    ![dot.gif][]\{  
 5![InBlock.gif][]        void Insert(OrderInfo order);  
 6![ExpandedSubBlockEnd.gif][]    \}  
 7![ExpandedBlockEnd.gif][]\}   
  8 ![None.gif][]  
  9 ![None.gif][] namespace  PetShop.BLL  
 10 ![ExpandedBlockStart.gif][]![ContractedBlock.gif][] ![dot.gif][] \{  
11![InBlock.gif][]    public class OrderSynchronous:IOrderStrategy  
12![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]    ![dot.gif][]\{  
13![InBlock.gif][]        private static readonly IOrder asynchOrder = QueueAccess.CreateOrder();  
14![InBlock.gif][]  
15![InBlock.gif][]        public void Insert(OrderInfo order)  
16![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]        ![dot.gif][]\{  
17![InBlock.gif][]            asynchOrder.Send(order);  
18![ExpandedSubBlockEnd.gif][]        \}  
19![ExpandedSubBlockEnd.gif][]    \}  
20![ExpandedBlockEnd.gif][]\}   
 21 ![None.gif][]  
 22 ![None.gif][] // ![dot.gif][]![dot.gif][]

从上面UML和代码就可以看出，订单策略接口下有两种实现，使用了抽象工厂模式来完成相应的订单策略对象的创建 。关于这点在后面消息工厂部分去介绍，这里不作讲解。  
  
2、消息接口IMessageing  
     在PetShop 4.0中，由于对订单处理使用了异步处理方式，在消息接口中仅定义了一个IOrder接口。IOrder接口的定义与MSMQ的实现是一致的，需要提供发送和接收操作。在Send方法中，参数为数据访问层的数据实体对象（OrderInfo），具体的实现则是用MSMQ的实现类(PetShop.MSMQMessaging.Order)去完成的。  
             ![petshop2.jpg][]           ![petshop3.jpg][]  
      MS的开发人员真的是什么都能想到，在消息接口的实现上考虑得很全面，为了避免将来的扩展会有其他的数据对象也使用到MSMQ；因此，在PetShop 4.0中的消息接口实现中，定义了一个队列的基类(PetShopQueue),实现了消息的发送(Send)和接收(Receive)方法的基本操作。代码如下：

1 ![None.gif][] namespace  PetShop.MSMQMessaging  
  2 ![ExpandedBlockStart.gif][]![ContractedBlock.gif][] ![dot.gif][] \{  
 3![InBlock.gif][]    public class PetShopQueue:IDisposable  
 4![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]    ![dot.gif][]\{  
 5![InBlock.gif][]        //指定消息队列事务的类型  
 6![InBlock.gif][]        protected MessageQueueTransactionType transactionType = MessageQueueTransactionType.Automatic;  
 7![InBlock.gif][]        protected MessageQueue queue;  //消息队列  
 8![InBlock.gif][]        protected TimeSpan timeout;    //时间间隔  
 9![InBlock.gif][]  
10![InBlock.gif][]        public PetShopQueue(string queuePath, int timeoutSeconds)  
11![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]        ![dot.gif][]\{  
12![InBlock.gif][]            queue = new MessageQueue(queuePath);  //根据传入quueuPath创建队列  
13![InBlock.gif][]            timeout = TimeSpan.FromSeconds(Convert.ToDouble(timeoutSeconds));  
14![InBlock.gif][]  
15![InBlock.gif][]            queue.DefaultPropertiesToSend.AttachSenderId = false;  
16![InBlock.gif][]            queue.DefaultPropertiesToSend.UseAuthentication = false;  
17![InBlock.gif][]            queue.DefaultPropertiesToSend.UseEncryption = false;  
18![InBlock.gif][]            queue.DefaultPropertiesToSend.AcknowledgeType = AcknowledgeTypes.None;  
19![InBlock.gif][]            queue.DefaultPropertiesToSend.UseJournalQueue = false;  
20![ExpandedSubBlockEnd.gif][]        \}  
21![InBlock.gif][]  
22![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]        /\*\*//// <summary>  
23![InBlock.gif][]        /// 接收消息方法  
24![ExpandedSubBlockEnd.gif][]        /// </summary>  
25![InBlock.gif][]        public virtual object Receive()  
26![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]        ![dot.gif][]\{  
27![InBlock.gif][]            try  
28![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]            ![dot.gif][]\{  
29![InBlock.gif][]                using (Message message = queue.Receive(timeout, transactionType))  
30![InBlock.gif][]                    return message;  
31![ExpandedSubBlockEnd.gif][]            \}  
32![InBlock.gif][]            catch (MessageQueueException mqex)  
33![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]            ![dot.gif][]\{  
34![InBlock.gif][]                if (mqex.MessageQueueErrorCode == MessageQueueErrorCode.IOTimeout)  
35![InBlock.gif][]                    throw new TimeoutException();  
36![InBlock.gif][]                throw;  
37![ExpandedSubBlockEnd.gif][]            \}  
38![ExpandedSubBlockEnd.gif][]        \}  
39![InBlock.gif][]  
40![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]        /\*\*//// <summary>  
41![InBlock.gif][]        /// 发送消息  
42![ExpandedSubBlockEnd.gif][]        /// </summary>  
43![InBlock.gif][]        public virtual void Send(object msg)  
44![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]        ![dot.gif][]\{  
45![InBlock.gif][]            queue.Send(msg, transactionType);  
46![ExpandedSubBlockEnd.gif][]        \}  
47![InBlock.gif][]  
48![ContractedSubBlock.gif][]![ExpandedSubBlockStart.gif][]        IDisposable 成员\#region IDisposable 成员  
49![InBlock.gif][]        public void Dispose()  
50![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]        ![dot.gif][]\{  
51![InBlock.gif][]            queue.Dispose();  //解放资源  
52![ExpandedSubBlockEnd.gif][]        \}  
53![ExpandedSubBlockEnd.gif][]        \#endregion  
54![ExpandedSubBlockEnd.gif][]    \}  
55![ExpandedBlockEnd.gif][]\}

MSMQ队列是一个可持久的队列，不会因用户不间断的下订单导致数据丢失。queue作为存放数据的队列，为消息队列(MessageQueue)类型，同时还为PetShopQueue设置了timeout值，后台处理应用程序(OrderProessor)会根据timeout的值定期扫描队列中的订单数量。  
  
3、消息工厂MessageFactory  
     可能是考虑到IOrder的实现会改变不同的策略吧，在PetShop里利用了抽象工厂模式，将IOrder对象的创建用了专门的工厂模块(MessageFactory)进行封装,定义如下：

1 ![None.gif][] namespace  PetShop.MessagingFactory  
  2 ![ExpandedBlockStart.gif][]![ContractedBlock.gif][] ![dot.gif][] \{  
 3![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]    /\*\*//// <summary>  
 4![InBlock.gif][]    /// This class is implemented following the Abstract Factory pattern to create the Order  
 5![InBlock.gif][]    /// Messaging implementation specified from the configuration file  
 6![ExpandedSubBlockEnd.gif][]    /// </summary>  
 7![InBlock.gif][]    public sealed class QueueAccess  
 8![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]    ![dot.gif][]\{  
 9![InBlock.gif][]        //<add key="OrderMessaging" value="PetShop.MSMQMessaging"/>  
10![InBlock.gif][]        private static readonly string path = "PetShop.MSMQMessaging";  
11![InBlock.gif][]  
12![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]        /\*\*//// <summary>  
13![InBlock.gif][]        /// 私有构造器,防止使用new创建对象实例  
14![ExpandedSubBlockEnd.gif][]        /// </summary>  
15![InBlock.gif][]        private QueueAccess()  
16![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]        ![dot.gif][]\{ \}  
17![InBlock.gif][]  
18![InBlock.gif][]        public static IOrder CreateOrder()  
19![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]        ![dot.gif][]\{  
20![InBlock.gif][]            string className = path \+ ".Order";  
21![InBlock.gif][]            return (IOrder)Assembly.Load(path).CreateInstance(className);  
22![ExpandedSubBlockEnd.gif][]        \}  
23![ExpandedSubBlockEnd.gif][]    \}  
24![ExpandedBlockEnd.gif][]\}

在QueueAccess类中，通过CreateOrder方法利用反射技术创建正确IOrder类型对象（实际也就是创建了一个接口的具体实现类的对象，应用了多态的原理和反射技术）。UML图下：  
                         ![petshop5.jpg][]  
     在PetShop4.0中，消息接口的具体实现是通过配置文件定义在web.config里：

![None.gif][] < add key = " OrderMessaging "  value = " PetShop.MSMQMessaging " />

这里我为了能够更直观的演示和介绍就把path固化定义了，如下：

![None.gif][] private   static   readonly   string  path  =   " PetShop.MSMQMessaging " ;

这里利用工厂模式来负责对象的创建，主要是方便业务逻辑层对定单处理策略的调用，如在PetShop.BLL模块中的OrderSynchronous类：

1 ![None.gif][] namespace  PetShop.BLL  
  2 ![ExpandedBlockStart.gif][]![ContractedBlock.gif][] ![dot.gif][] \{  
 3![InBlock.gif][]    public class OrderSynchronous:IOrderStrategy  
 4![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]    ![dot.gif][]\{  
 5![InBlock.gif][]        private static readonly IOrder asynchOrder = QueueAccess.CreateOrder();  
 6![InBlock.gif][]  
 7![InBlock.gif][]        public void Insert(OrderInfo order)  
 8![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]        ![dot.gif][]\{  
 9![InBlock.gif][]            asynchOrder.Send(order);  
10![ExpandedSubBlockEnd.gif][]        \}  
11![ExpandedSubBlockEnd.gif][]    \}  
12![ExpandedBlockEnd.gif][]\}

这样一但IOrder接口的实现发生了变化，此时就只需要修改配置文件就OK，整个系统就显得很灵活，稳定。  
  
4、MSMQ实现MSMQMessaging  
    在PetShop.MSMQMessaging模块中，订单对象实现了消息接口（IMessaging）模块中的IOrder，同时还继承了基类PetShopQueue。定义如下：

1 ![None.gif][] namespace  PetShop.MSMQMessaging  
  2 ![ExpandedBlockStart.gif][]![ContractedBlock.gif][] ![dot.gif][] \{  
 3![InBlock.gif][]    public class Order:PetShopQueue,IOrder  
 4![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]    ![dot.gif][]\{  
 5![InBlock.gif][]        private static readonly string queuePath = ConfigurationManager.AppSettings\["OrderQueuePath"\];  
 6![InBlock.gif][]        private static int queueTimeout = 20;  //20秒为超时  
 7![InBlock.gif][]  
 8![InBlock.gif][]        public Order()  
 9![InBlock.gif][]            : base(queuePath, queueTimeout)  
10![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]        ![dot.gif][]\{  
11![InBlock.gif][]            queue.Formatter = new BinaryMessageFormatter();  
12![InBlock.gif][]            Console.WriteLine(queuePath);  
13![ExpandedSubBlockEnd.gif][]        \}  
14![InBlock.gif][]  
15![InBlock.gif][]        public new OrderInfo Receive()  
16![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]        ![dot.gif][]\{  
17![InBlock.gif][]            //该方法会应用在分布式事务中，故而设置为Automatic的事务类型。  
18![InBlock.gif][]            base.transactionType = MessageQueueTransactionType.Automatic;  
19![InBlock.gif][]            return (OrderInfo)((Message)base.Receive()).Body;  
20![ExpandedSubBlockEnd.gif][]        \}  
21![InBlock.gif][]  
22![InBlock.gif][]        public OrderInfo Receive(int timeout)  
23![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]        ![dot.gif][]\{  
24![InBlock.gif][]            base.timeout = TimeSpan.FromSeconds(Convert.ToDouble(timeout));  
25![InBlock.gif][]            return Receive();  
26![ExpandedSubBlockEnd.gif][]        \}  
27![InBlock.gif][]  
28![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]        /\*\*//// <summary>  
29![InBlock.gif][]        /// 异步发送定单到消息队列  
30![InBlock.gif][]        /// </summary>  
31![ExpandedSubBlockEnd.gif][]        /// <param name="orderMessage"></param>  
32![InBlock.gif][]        public void Send(OrderInfo orderMessage)  
33![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]        ![dot.gif][]\{  
34![InBlock.gif][]            //该方法不会用于分布式事务中，故而设置为Single的事务类型。  
35![InBlock.gif][]            base.transactionType = MessageQueueTransactionType.Single;  
36![InBlock.gif][]            base.Send(orderMessage);  
37![ExpandedSubBlockEnd.gif][]        \}  
38![ExpandedSubBlockEnd.gif][]    \}  
39![ExpandedBlockEnd.gif][]\}

UML草图：  
                                      ![petshop4.jpg][]  
     这里需要注意的是，Order类既继承了基类PetShopQueue，同时还实现了接口IOrder，而在消息接口和基类中所定义的接收消息(Receive)方法在方法的签名上是相同的。所以在Order的Receive方法实现中，必须使用new而非override关键字来重写父类PetShopQueue的Receive虚方法。此时Order类的Receive方法代表两个含义，一是实现了消息接口IOrder中的Receive方法；二则是利用了new关键字重写了父类PetShopQueue的Receive虚方法。  
     在PetShop4.0中，将面向对象的知识点应用得非常精妙，如上分析，此时我门可以怎么来调用Order呢？是这样吗？

1 ![None.gif][] // 1、使用基类PetShopQueue   
 2 ![None.gif][] PetShopQueue order  =   new  Order();  
 3 ![None.gif][]order.Receive();  
 4 ![None.gif][]  
 5 ![None.gif][] // 2、使用消息接口IOrder   
 6 ![None.gif][] IOrder  order  =   new  Order();  
 7 ![None.gif][]order.Receive();

根据多态原理，上面这两种实现都是正确的，那我们那取谁舍谁呢？在PetShop4.0中正确的调用是第2种方法，这种调用方法也更符合“面向接口设计”的原则。----详细请查看消息工厂MessageFactory部分的介绍。  
  
5、后台处理应用程序OrderProessor  
      前面一系列的操作，最终都会走向到这里，这里实现了最终的插入数据库的操作。在PetShop 4.0中OrderProessor是一个控制台应用程序，根据需求也可以将其设计为Windows Service。他完成的操作就是接收消息队列里的订单数据，将其插入到数据库。在OrderProessor里使用了多线程技术，监视队列里的订单信息，定期的将其处理。在主方法Main方法中用于控制线程序，核心的执行任何则委托给ProcessOrders方法去实现。

1 ![None.gif][] private   static   void  ProcessOrders()  
  2 ![ExpandedBlockStart.gif][]![ContractedBlock.gif][] ![dot.gif][] \{  
 3![InBlock.gif][]    TimeSpan tsTimeout = TimeSpan.FromSeconds(Convert.ToDouble(transactionTimeout \* batchSize));  
 4![InBlock.gif][]    Order order = new Order();  //逻辑层的PetShop.BLL.Order  
 5![InBlock.gif][]    while (true)  
 6![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]    ![dot.gif][]\{  
 7![InBlock.gif][]        TimeSpan datatimeStarting = new TimeSpan(DateTime.Now.Ticks);  
 8![InBlock.gif][]        double elapsedTime = 0;  
 9![InBlock.gif][]  
10![InBlock.gif][]        int processedItems = 0;  
11![InBlock.gif][]        ArrayList queueOrders = new ArrayList();  
12![InBlock.gif][]  
13![InBlock.gif][]        //首先验证事务  
14![InBlock.gif][]        using (TransactionScope ts = new TransactionScope(TransactionScopeOption.Required, tsTimeout))  
15![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]        ![dot.gif][]\{  
16![InBlock.gif][]            //从队列中检索订单  
17![InBlock.gif][]            for (int i = 0; i < batchSize; i\++)  
18![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]            ![dot.gif][]\{  
19![InBlock.gif][]                try  
20![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]                ![dot.gif][]\{  
21![InBlock.gif][]                    //在一定时间 类接收队列的订单  
22![InBlock.gif][]                    if ((elapsedTime \+ queueTimeout \+ transactionTimeout) < tsTimeout.TotalSeconds)  
23![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]                    ![dot.gif][]\{  
24![InBlock.gif][]                        queueOrders.Add(order.ReceiveFromQueue(queueTimeout));  
25![ExpandedSubBlockEnd.gif][]                    \}  
26![InBlock.gif][]                    else  
27![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]                    ![dot.gif][]\{  
28![InBlock.gif][]                        i = batchSize;  // 结束循环  
29![ExpandedSubBlockEnd.gif][]                    \}  
30![InBlock.gif][]                    elapsedTime = new TimeSpan(DateTime.Now.Ticks).TotalSeconds \- datatimeStarting.TotalSeconds;  
31![ExpandedSubBlockEnd.gif][]                \}  
32![InBlock.gif][]                catch (TimeoutException)  
33![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]                ![dot.gif][]\{  
34![InBlock.gif][]                    //没有可以等待的消息也结束循环  
35![InBlock.gif][]                    i = batchSize;  
36![ExpandedSubBlockEnd.gif][]                \}  
37![ExpandedSubBlockEnd.gif][]            \}  
38![InBlock.gif][]  
39![InBlock.gif][]            //处理队列的订单  
40![InBlock.gif][]            for (int k = 0; k < queueOrders.Count; k\++)  
41![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]            ![dot.gif][]\{  
42![InBlock.gif][]                order.Insert((OrderInfo)queueOrders\[k\]);  
43![InBlock.gif][]                processedItems\++;  
44![InBlock.gif][]                totalOrdersProcessed\++;  
45![ExpandedSubBlockEnd.gif][]            \}  
46![InBlock.gif][]            //处理完毕或者是超时  
47![InBlock.gif][]            ts.Complete();  
48![ExpandedSubBlockEnd.gif][]        \}  
49![InBlock.gif][]        Console.WriteLine("(Thread Id " \+ Thread.CurrentThread.ManagedThreadId \+ ") batch finished, " \+ processedItems \+ " items, in " \+ elapsedTime.ToString() \+ " seconds.");  
50![ExpandedSubBlockEnd.gif][]    \}  
51![ExpandedBlockEnd.gif][]\}

ProcessOrders方法首先通过业务逻辑层 PetShop.BLL.Order类的方法ReceiveFromQueue去获取消息队列中的订单数据，并将其放入一个ArrayList对象，然后将其插入到数据库。 OrderProcessor的完整代码定义如下：

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][] Code   
  1![None.gif][]using System;  
  2![None.gif][]using System.Collections.Generic;  
  3![None.gif][]using System.Text;  
  4![None.gif][]using System.Configuration;  
  5![None.gif][]using System.Threading;  
  6![None.gif][]using PetShop.BLL;  
  7![None.gif][]using System.Collections;  
  8![None.gif][]using System.Transactions;  
  9![None.gif][]using PetShop.Model;  
 10![None.gif][]  
 11![None.gif][]namespace PetShop.OrderProcessor  
 12![ExpandedBlockStart.gif][]![ContractedBlock.gif][]![dot.gif][]\{  
 13![InBlock.gif][]    class Program  
 14![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]    ![dot.gif][]\{  
 15![InBlock.gif][]        private static int transactionTimeout = int.Parse(ConfigurationManager.AppSettings\["TransactionTimeout"\]);  
 16![InBlock.gif][]        private static int queueTimeout = int.Parse(ConfigurationManager.AppSettings\["QueueTimeout"\]);  
 17![InBlock.gif][]        private static int batchSize = int.Parse(ConfigurationManager.AppSettings\["BatchSize"\]);  
 18![InBlock.gif][]        private static int threadCount = int.Parse(ConfigurationManager.AppSettings\["ThreadCount"\]);  
 19![InBlock.gif][]        private static int totalOrdersProcessed = 0;  
 20![InBlock.gif][]  
 21![InBlock.gif][]        static void Main(string\[\] args)  
 22![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]        ![dot.gif][]\{  
 23![InBlock.gif][]            Thread workTicketThread;  
 24![InBlock.gif][]            Thread\[\] workerThreads = new Thread\[threadCount\];  
 25![InBlock.gif][]  
 26![InBlock.gif][]            for (int i = 0; i < threadCount; i\++)  
 27![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]            ![dot.gif][]\{  
 28![InBlock.gif][]                workTicketThread = new Thread(new ThreadStart(ProcessOrders));  
 29![InBlock.gif][]                //指示线呈是否为一后台线程  
 30![InBlock.gif][]                workTicketThread.IsBackground = true;  
 31![InBlock.gif][]                workTicketThread.SetApartmentState(ApartmentState.STA);  
 32![InBlock.gif][]  
 33![InBlock.gif][]                workTicketThread.Start();  
 34![InBlock.gif][]                workerThreads\[i\] = workTicketThread;  
 35![ExpandedSubBlockEnd.gif][]            \}  
 36![InBlock.gif][]  
 37![InBlock.gif][]            Console.WriteLine("开始处理,按任意键停止.");  
 38![InBlock.gif][]            Console.ReadLine();  
 39![InBlock.gif][]            Console.WriteLine("正在终止线程,请等待![dot.gif][]![dot.gif][]");  
 40![InBlock.gif][]  
 41![InBlock.gif][]            //终止所以线程  
 42![InBlock.gif][]            for (int i = 0; i < workerThreads.Length; i\++)  
 43![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]            ![dot.gif][]\{  
 44![InBlock.gif][]                workerThreads\[i\].Abort();  
 45![ExpandedSubBlockEnd.gif][]            \}  
 46![InBlock.gif][]  
 47![InBlock.gif][]            Console.WriteLine();  
 48![InBlock.gif][]            Console.WriteLine(totalOrdersProcessed \+ " 张订单已经处理.");  
 49![InBlock.gif][]            Console.WriteLine("已终止处理.按任意键退出![dot.gif][]![dot.gif][]");  
 50![InBlock.gif][]            Console.ReadLine();  
 51![ExpandedSubBlockEnd.gif][]        \}  
 52![InBlock.gif][]  
 53![InBlock.gif][]        private static void ProcessOrders()  
 54![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]        ![dot.gif][]\{  
 55![InBlock.gif][]            TimeSpan tsTimeout = TimeSpan.FromSeconds(Convert.ToDouble(transactionTimeout \* batchSize));  
 56![InBlock.gif][]            Order order = new Order();  //逻辑层的PetShop.BLL.Order  
 57![InBlock.gif][]            while (true)  
 58![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]            ![dot.gif][]\{  
 59![InBlock.gif][]                TimeSpan datatimeStarting = new TimeSpan(DateTime.Now.Ticks);  
 60![InBlock.gif][]                double elapsedTime = 0;  
 61![InBlock.gif][]  
 62![InBlock.gif][]                int processedItems = 0;  
 63![InBlock.gif][]                ArrayList queueOrders = new ArrayList();  
 64![InBlock.gif][]  
 65![InBlock.gif][]                //首先验证事务  
 66![InBlock.gif][]                using (TransactionScope ts = new TransactionScope(TransactionScopeOption.Required, tsTimeout))  
 67![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]                ![dot.gif][]\{  
 68![InBlock.gif][]                    //从队列中检索订单  
 69![InBlock.gif][]                    for (int i = 0; i < batchSize; i\++)  
 70![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]                    ![dot.gif][]\{  
 71![InBlock.gif][]                        try  
 72![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]                        ![dot.gif][]\{  
 73![InBlock.gif][]                            //在一定时间 类接收队列的订单  
 74![InBlock.gif][]                            if ((elapsedTime \+ queueTimeout \+ transactionTimeout) < tsTimeout.TotalSeconds)  
 75![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]                            ![dot.gif][]\{  
 76![InBlock.gif][]                                queueOrders.Add(order.ReceiveFromQueue(queueTimeout));  
 77![ExpandedSubBlockEnd.gif][]                            \}  
 78![InBlock.gif][]                            else  
 79![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]                            ![dot.gif][]\{  
 80![InBlock.gif][]                                i = batchSize;  // 结束循环  
 81![ExpandedSubBlockEnd.gif][]                            \}  
 82![InBlock.gif][]                            elapsedTime = new TimeSpan(DateTime.Now.Ticks).TotalSeconds \- datatimeStarting.TotalSeconds;  
 83![ExpandedSubBlockEnd.gif][]                        \}  
 84![InBlock.gif][]                        catch (TimeoutException)  
 85![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]                        ![dot.gif][]\{  
 86![InBlock.gif][]                            //没有可以等待的消息也结束循环  
 87![InBlock.gif][]                            i = batchSize;  
 88![ExpandedSubBlockEnd.gif][]                        \}  
 89![ExpandedSubBlockEnd.gif][]                    \}  
 90![InBlock.gif][]  
 91![InBlock.gif][]                    //处理队列的订单  
 92![InBlock.gif][]                    for (int k = 0; k < queueOrders.Count; k\++)  
 93![ExpandedSubBlockStart.gif][]![ContractedSubBlock.gif][]                    ![dot.gif][]\{  
 94![InBlock.gif][]                        order.Insert((OrderInfo)queueOrders\[k\]);  
 95![InBlock.gif][]                        processedItems\++;  
 96![InBlock.gif][]                        totalOrdersProcessed\++;  
 97![ExpandedSubBlockEnd.gif][]                    \}  
 98![InBlock.gif][]                    //处理完毕或者是超时  
 99![InBlock.gif][]                    ts.Complete();  
100![ExpandedSubBlockEnd.gif][]                \}  
101![InBlock.gif][]                Console.WriteLine("(Thread Id " \+ Thread.CurrentThread.ManagedThreadId \+ ") batch finished, " \+ processedItems \+ " items, in " \+ elapsedTime.ToString() \+ " seconds.");  
102![ExpandedSubBlockEnd.gif][]            \}  
103![ExpandedSubBlockEnd.gif][]        \}  
104![ExpandedSubBlockEnd.gif][]    \}  
105![ExpandedBlockEnd.gif][]\}  
106![None.gif][]

MSMQ技术除了用于异步处理之外，还可作为一种分布式处理技术应用。关于使用MSMQ进行分布式处理相关的内容，本人能力有限，还请大家查看相关的资料进行了解。  
      本文介绍了MSMQ和多线程以及对PetShop 4.0中对MSMQ的应用进行了分析，结合之前我写的两篇关于MSMQ的相关知识点的介绍文章，对MSMQ算是建立了一个全面的认识，希望这三篇文章对大家学习MSMQ有所帮助。

**[文中示例代码下载][Link 1]**

转载于:https://www.cnblogs.com/gooddasenlin/archive/2011/04/22/2025194.html

[None.gif]: /images/20211210/6e3c9a3bf8064478a3c01447aaf6c4bb.png
[ExpandedBlockStart.gif]: https://www.cnblogs.com/Images/OutliningIndicators/ExpandedBlockStart.gif
[ContractedBlock.gif]: https://www.cnblogs.com/Images/OutliningIndicators/ContractedBlock.gif
[dot.gif]: /images/20211210/dee3c167a2124119a9be333ac89f2ecc.png
[InBlock.gif]: /images/20211210/fbc7faf7820b4a9caa3678fc314cfc67.png
[ExpandedBlockEnd.gif]: /images/20211210/4aa81c7c093b422fbd9143ccaa183b0c.png
[ExpandedSubBlockStart.gif]: /images/20211210/4a03e47885a446d2a1c6b9c1bb7228a5.png
[ContractedSubBlock.gif]: /images/20211210/fd9d1154c6d448d1bcb1236192e3d676.png
[ExpandedSubBlockEnd.gif]: /images/20211210/257b921c04cf4b74b97e883836c07d8f.png
[ASP.NET_MSMQ_]: http://www.cnblogs.com/beniao/archive/2008/06/28/1230311.html
[msmq9.jpg]: /images/20211210/1b3105b9d39549a79180c2ebe74596cb.png
[msmq10.jpg]: /images/20211210/9e5a2b08922f445ca0985c93677dbc88.png
[tongue_smile.gif]: /images/20211210/ab764d0e41024e2dbe1cf77b04f55069.png
[petshop0.jpg]: /images/20211210/3224a98198b34883a8cd63954fe86088.png
[petshop1.jpg]: /images/20211210/7e7a8dd982984cd0aa44fcbba2aa051e.png
[petshop2.jpg]: https://images.cnblogs.com/cnblogs_com/beniao/petshop2.jpg
[petshop3.jpg]: /images/20211210/6c4744c97e0d4bb49c508ff836710179.png
[petshop5.jpg]: https://images.cnblogs.com/cnblogs_com/beniao/petshop5.jpg
[petshop4.jpg]: https://images.cnblogs.com/cnblogs_com/beniao/petshop4.jpg
[Link 1]: http://files.cnblogs.com/beniao/MSMQ.Demo.rar