petshop详解之三：PetShop数据访问层之消息处理

三、PetShop数据访问层之消息处理
在进行系统设计时&＃xff0c;除了对安全、事务等问题给与足够的重视外&＃xff0c;性能也是一个不可避免的问题所在&＃xff0c;尤其是一个B/S结构的软件系统&＃xff0c;必须充分地考虑访问量、数据流量、服务器负荷的问题。解决性能的瓶颈&＃xff0c;除了对硬件系统进行升级外&＃xff0c;软件设计的合理性尤为重要。
在前面我曾提到&＃xff0c;分层式结构设计可能会在一定程度上影响数据访问的性能&＃xff0c;然而与它给设计人员带来的好处相比&＃xff0c;几乎可以忽略。要提供整个系统的性能&＃xff0c;还可以从数据库的优化着手&＃xff0c;例如连接池的使用、建立索引、优化查询策略等等&＃xff0c;例如在PetShop中就利用了数据库的Cache&＃xff0c;对于数据量较大的订单数据&＃xff0c;则利用分库的方式为其单独建立了Order和Inventory数据库。而在软件设计上&＃xff0c;比较有用的方式是利用多线程与异步处理方式。
在PetShop4.0中&＃xff0c;使用了Microsoft Messaging Queue(MSMQ)技术来完成异步处理&＃xff0c;利用消息队列临时存放要插入的数据&＃xff0c;使得数据访问因为不需要访问数据库从而提供了访问性能&＃xff0c;至于队列中的数据&＃xff0c;则等待系统空闲的时候再进行处理&＃xff0c;将其最终插入到数据库中。
PetShop4.0中的消息处理&＃xff0c;主要分为如下几部分&＃xff1a;消息接口IMessaging、消息工厂MessagingFactory、MSMQ实现MSMQMessaging以及数据后台处理应用程序OrderProcessor。
从模块化分上&＃xff0c;PetShop自始自终地履行了“面向接口设计”的原则&＃xff0c;将消息处理的接口与实现分开&＃xff0c;并通过工厂模式封装消息实现对象的创建&＃xff0c;以达到松散耦合的目的。
由于在PetShop中仅对订单的处理使用了异步处理方式&＃xff0c;因此在消息接口IMessaging中&＃xff0c;仅定义了一个IOrder接口&＃xff0c;其类图如下&＃xff1a;
 
在对消息接口的实现中&＃xff0c;考虑到未来的扩展中会有其他的数据对象会使用MSMQ&＃xff0c;因此定义了一个Queue的基类&＃xff0c;实现消息Receive和Send的基本操作&＃xff1a;
public virtual object Receive()
{
      try
{
          using (Message message &＃61; queue.Receive(timeout, transactionType))
             return message;
      }
      catch (MessageQueueException mqex)
{
          if (mqex.MessageQueueErrorCode &＃61;&＃61; MessageQueueErrorCode.IOTimeout)
             throw new TimeoutException();
                throw;
      }
}
public virtual void Send(object msg)
{
      queue.Send(msg, transactionType);
}
其中queue对象是System.Messaging.MessageQueue类型&＃xff0c;作为存放数据的队列。MSMQ队列是一个可持久的队列&＃xff0c;因此不必担心用户不间断地下订单会导致订单数据的丢失。在PetShopQueue设置了timeout值&＃xff0c;OrderProcessor会根据timeout值定期扫描队列中的订单数据。
MSMQMessaging模块中&＃xff0c;Order对象实现了IMessaging模块中定义的接口IOrder&＃xff0c;同时它还继承了基类PetShopQueue&＃xff0c;其定义如下&＃xff1a;
public class Order:PetShopQueue, PetShop.IMessaging.IOrder
方法的实现代码如下&＃xff1a;
    public new OrderInfo Receive()
    {
        // This method involves in distributed transaction and need Automatic Transaction type
        base.transactionType &＃61; MessageQueueTransactionType.Automatic;
        return (OrderInfo)((Message)base.Receive()).Body;
    }

    public OrderInfo Receive(int timeout)
    {
        base.timeout &＃61; TimeSpan.FromSeconds(Convert.ToDouble(timeout));
        return Receive();
    }

    public void Send(OrderInfo orderMessage)
    {
        // This method does not involve in distributed transaction and optimizes performance using Single type
        base.transactionType &＃61; MessageQueueTransactionType.Single;
        base.Send(orderMessage);
    }
所以&＃xff0c;最后的类图应该如下&＃xff1a;
 
注意在Order类的Receive()方法中&＃xff0c;是用new关键字而不是override关键字来重写其父类PetShopQueue的Receive()虚方法。因此&＃xff0c;如果是实例化如下的对象&＃xff0c;将会调用PetShopQueue的Receive()方法&＃xff0c;而不是子类Order的Receive()方法&＃xff1a;
PetShopQueue queue &＃61; new Order();
queue.Receive();
从设计上来看&＃xff0c;由于PetShop采用“面向接口设计”的原则&＃xff0c;如果我们要创建Order对象&＃xff0c;应该采用如下的方式&＃xff1a;
IOrder order &＃61; new Order();
order.Receive();
考虑到IOrder的实现有可能的变化&＃xff0c;PetShop仍然利用了工厂模式&＃xff0c;将IOrder对象的创建用专门的工厂模块进行了封装&＃xff1a;
 
在类QueueAccess中&＃xff0c;通过CreateOrder()方法利用反射技术创建正确的IOrder类型对象&＃xff1a;
    public static PetShop.IMessaging.IOrder CreateOrder()
    {
        string className &＃61; path &＃43; ".Order";
        return PetShop.IMessaging.IOrder)Assembly.Load(path).CreateInstance(className);
    }
path的值通过配置文件获取&＃xff1a;
private static readonly string path &＃61; ConfigurationManager.AppSettings["OrderMessaging"];
而配置文件中&＃xff0c;OrderMessaging的值设置如下&＃xff1a;

之所以利用工厂模式来负责对象的创建&＃xff0c;是便于在业务层中对其调用&＃xff0c;例如在BLL模块中OrderAsynchronous类&＃xff1a;
public class OrderAsynchronous : IOrderStrategy
{       
    private static readonly PetShop.IMessaging.IOrder asynchOrder &＃61; PetShop.MessagingFactory.QueueAccess.CreateOrder();
    public void Insert(PetShop.Model.OrderInfo order)
{
        asynchOrder.Send(order);
    }
}
一旦IOrder接口的实现发生变化&＃xff0c;这种实现方式就可以使得客户仅需要修改配置文件&＃xff0c;而不需要修改代码&＃xff0c;如此就可以避免程序集的重新编译和部署&＃xff0c;使得系统能够灵活应对需求的改变。例如定义一个实现IOrder接口的SpecialOrder&＃xff0c;则可以新增一个模块&＃xff0c;如PetShop.SpecialMSMQMessaging&＃xff0c;而类名则仍然为Order&＃xff0c;那么此时我们仅需要修改配置文件中OrderMessaging的值即可&＃xff1a;

OrderProcessor是一个控制台应用程序&＃xff0c;不过可以根据需求将其设计为Windows Service。它的目的就是接收消息队列中的订单数据&＃xff0c;然后将其插入到Order和Inventory数据库中。它利用了多线程技术&＃xff0c;以达到提高系统性能的目的。
在OrderProcessor应用程序中&＃xff0c;主函数Main用于控制线程&＃xff0c;而核心的执行任务则由方法ProcessOrders()实现&＃xff1a;
    private static void ProcessOrders()
    {
        // the transaction timeout should be long enough to handle all of orders in the batch
        TimeSpan tsTimeout &＃61; TimeSpan.FromSeconds(Convert.ToDouble(transactionTimeout * batchSize));

        Order order &＃61; new Order();
        while (true)
        {
            // queue timeout variables
            TimeSpan datetimeStarting &＃61; new TimeSpan(DateTime.Now.Ticks);
            double elapsedTime &＃61; 0;

            int processedItems &＃61; 0;

            ArrayList queueOrders &＃61; new ArrayList();

            using (TransactionScope ts &＃61; new TransactionScope(TransactionScopeOption.Required, tsTimeout))
            {
                // Receive the orders from the queue
                for (int j &＃61; 0; j                 {
                    try
                    {
                        //only receive more queued orders if there is enough time
                        if ((elapsedTime &＃43; queueTimeout &＃43; transactionTimeout)                         {
                            queueOrders.Add(order.ReceiveFromQueue(queueTimeout));
                        }
                        else
                        {
                            j &＃61; batchSize;   // exit loop
                        }

                        //update elapsed time
                        elapsedTime &＃61; new TimeSpan(DateTime.Now.Ticks).TotalSeconds - datetimeStarting.TotalSeconds;
                    }
                    catch (TimeoutException)
                    {
                        //exit loop because no more messages are waiting
                        j &＃61; batchSize;
                    }
                }
                //process the queued orders
                for (int k &＃61; 0; k                 {
                    order.Insert((OrderInfo)queueOrders[k]);
                    processedItems&＃43;&＃43;;
                    totalOrdersProcessed&＃43;&＃43;;
                }

                //batch complete or MSMQ receive timed out
                ts.Complete();
            }

            Console.WriteLine("(Thread Id " &＃43; Thread.CurrentThread.ManagedThreadId &＃43; ") batch finished, " &＃43; processedItems &＃43; " items, in " &＃43; elapsedTime.ToString() &＃43; " seconds.");
        }
    }
首先&＃xff0c;它会通过PetShop.BLL.Order类的公共方法ReceiveFromQueue()来获取消息队列中的订单数据&＃xff0c;并将其放入到一个ArrayList对象中&＃xff0c;然而再调用PetShop.BLL.Order类的Insert方法将其插入到Order和Inventory数据库中。
在PetShop.BLL.Order类中&＃xff0c;并不是直接执行插入订单的操作&＃xff0c;而是调用了IOrderStrategy接口的Insert()方法&＃xff1a;
public void Insert(OrderInfo order)
{
    // Call credit card procesor
    ProcessCreditCard(order);

    // Insert the order (a)synchrounously based on configuration
    orderInsertStrategy.Insert(order);
}
在这里&＃xff0c;运用了一个策略模式&＃xff0c;类图如下所示&＃xff1a;
 
在PetShop.BLL.Order类中&＃xff0c;仍然利用配置文件来动态创建IOrderStategy对象&＃xff1a;
private static readonly PetShop.IBLLStrategy.IOrderStrategy orderInsertStrategy &＃61; LoadInsertStrategy();
private static PetShop.IBLLStrategy.IOrderStrategy LoadInsertStrategy()
{
    // Look up which strategy to use from config file
    string path &＃61; ConfigurationManager.AppSettings["OrderStrategyAssembly"];
    string className &＃61; ConfigurationManager.AppSettings["OrderStrategyClass"];

    // Using the evidence given in the config file load the appropriate assembly and class
    return (PetShop.IBLLStrategy.IOrderStrategy)Assembly.Load(path).CreateInstance(className);
}
由于OrderProcessor是一个单独的应用程序&＃xff0c;因此它使用的配置文件与PetShop不同&＃xff0c;是存放在应用程序的App.config文件中&＃xff0c;在该文件中&＃xff0c;对IOrderStategy的配置为&＃xff1a;


因此&＃xff0c;以异步方式插入订单的流程如下图所示&＃xff1a;
 
Microsoft Messaging Queue(MSMQ)技术除用于异步处理以外&＃xff0c;它主要还是一种分布式处理技术。分布式处理中&＃xff0c;一个重要的技术要素就是有关消息的处理&＃xff0c;而在System.Messaging命名空间中&＃xff0c;已经提供了Message类&＃xff0c;可以用于承载消息的传递&＃xff0c;前提上消息的发送方与接收方在数据定义上应有统一的接口规范。
MSMQ在分布式处理的运用&＃xff0c;在我参与的项目中已经有了实现。在为一个汽车制造商开发一个大型系统时&＃xff0c;分销商Dealer作为.Net客户端&＃xff0c;需要将数据传递到管理中心&＃xff0c;并且该数据将被Oracle的EBS(E-Business System)使用。由于分销商管理系统&＃xff08;DMS&＃xff09;采用的是C/S结构&＃xff0c;数据库为SQL Server&＃xff0c;而汽车制造商管理中心的EBS数据库为Oracle。这里就涉及到两个系统之间数据的传递。
实现架构如下&＃xff1a;

     首先Dealer的数据通过MSMQ传递到MSMQ Server&＃xff0c;此时可以将数据插入到SQL Server数据库中&＃xff0c;同时利用FTP将数据传送到专门的文件服务器上。然后利用IBM的EAI技术&＃xff08;企业应用集成&＃xff0c;Enterprise Application Itegration&＃xff09;定期将文件服务器中的文件&＃xff0c;利用接口规范写入到EAI数据库服务器中&＃xff0c;并最终写道EBS的Oracle数据库中。
上述架构是一个典型的分布式处理结构&＃xff0c;而技术实现的核心就是MSMQ和EAI。由于我们已经定义了统一的接口规范&＃xff0c;在通过消息队列形成文件后&＃xff0c;此时的数据就已经与平台无关了&＃xff0c;使得在.Net平台下的分销商管理系统能够与Oracle的EBS集成起来&＃xff0c;完成数据的处理。