《解剖PetShop》之三：PetShop数据访问层之消息处理
三、petshop数据访问层之消息处理
在进行系统设计时，除了对安全、事务等问题给与足够的重视外，性能也是一个不可避免的问题所在，尤其是一个b/s结构的软件系统，必须充分地考虑访问量、数据流量、服务器负荷的问题。解决性能的瓶颈，除了对硬件系统进行升级外，软件设计的合理性尤为重要。
在前面我曾提到，分层式结构设计可能会在一定程度上影响数据访问的性能，然而与它给设计人员带来的好处相比，几乎可以忽略。要提供整个系统的性能，还可以从数据库的优化着手，例如连接池的使用、建立索引、优化查询策略等等，例如在petshop中就利用了数据库的cache，对于数据量较大的订单数据，则利用分库的方式为其单独建立了order和inventory数据库。而在软件设计上，比较有用的方式是利用多线程与异步处理方式。
在petshop4.0中，使用了microsoft messaging queue(msmq)技术来完成异步处理，利用消息队列临时存放要插入的数据，使得数据访问因为不需要访问数据库从而提供了访问性能，至于队列中的数据，则等待系统空闲的时候再进行处理，将其最终插入到数据库中。
petshop4.0中的消息处理，主要分为如下几部分：消息接口imessaging、消息工厂messagingfactory、msmq实现msmqmessaging以及数据后台处理应用程序orderprocessor。
从模块化分上，petshop自始自终地履行了“面向接口设计”的原则，将消息处理的接口与实现分开，并通过工厂模式封装消息实现对象的创建，以达到松散耦合的目的。
由于在petshop中仅对订单的处理使用了异步处理方式，因此在消息接口imessaging中，仅定义了一个iorder接口，其类图如下：
在对消息接口的实现中，考虑到未来的扩展中会有其他的数据对象会使用msmq，因此定义了一个queue的基类，实现消息receive和send的基本操作：
public virtual object receive()
{
try
{
using (message message = queue.receive(timeout, transactiontype))
return message;
}
catch (messagequeueexception mqex)
{
if (mqex.messagequeueerrorcode == messagequeueerrorcode.iotimeout)
throw new timeoutexception();
throw;
}
}
public virtual void send(object msg)
{
queue.send(msg, transactiontype);
}
其中queue对象是system.messaging.messagequeue类型，作为存放数据的队列。msmq队列是一个可持久的队列，因此不必担心用户不间断地下订单会导致订单数据的丢失。在petshopqueue设置了timeout值，orderprocessor会根据timeout值定期扫描队列中的订单数据。
msmqmessaging模块中，order对象实现了imessaging模块中定义的接口iorder，同时它还继承了基类petshopqueue，其定义如下：
public class order:petshopqueue, petshop.imessaging.iorder
方法的实现代码如下：
public new orderinfo receive()
{
// this method involves in distributed transaction and need automatic transaction type
base.transactiontype = messagequeuetransactiontype.automatic;
return (orderinfo)((message)base.receive()).body;
}
public orderinfo receive(int timeout)
{
base.timeout = 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 = messagequeuetransactiontype.single;
base.send(ordermessage);
}
所以，最后的类图应该如下：
注意在order类的receive()方法中，是用new关键字而不是override关键字来重写其父类petshopqueue的receive()虚方法。因此，如果是实例化如下的对象，将会调用petshopqueue的receive()方法，而不是子类order的receive()方法：
petshopqueue queue = new order(); queue.receive();
从设计上来看，由于petshop采用“面向接口设计”的原则，如果我们要创建order对象，应该采用如下的方式：
iorder order = new order(); order.receive();
考虑到iorder的实现有可能的变化，petshop仍然利用了工厂模式，将iorder对象的创建用专门的工厂模块进行了封装：
在类queueaccess中，通过createorder()方法利用反射技术创建正确的iorder类型对象：
public static petshop.imessaging.iorder createorder()
{
string classname = path + ".order";
return petshop.imessaging.iorder)assembly.load(path).createinstance(classname);
}
path的值通过配置文件获取：
private static readonly string path = configurationmanager.appsettings["ordermessaging"];
而配置文件中，ordermessaging的值设置如下：
<add key="ordermessaging" value="petshop.msmqmessaging"/>
之所以利用工厂模式来负责对象的创建，是便于在业务层中对其调用，例如在bll模块中orderasynchronous类：
public class orderasynchronous : iorderstrategy
{
private static readonly petshop.imessaging.iorder asynchorder = petshop.messagingfactory.queueaccess.createorder();
public void insert(petshop.model.orderinfo order)
{
asynchorder.send(order);
}
}
一旦iorder接口的实现发生变化，这种实现方式就可以使得客户仅需要修改配置文件，而不需要修改代码，如此就可以避免程序集的重新编译和部署，使得系统能够灵活应对需求的改变。例如定义一个实现iorder接口的specialorder，则可以新增一个模块，如petshop.specialmsmqmessaging，而类名则仍然为order，那么此时我们仅需要修改配置文件中ordermessaging的值即可：
<add key="ordermessaging" value="petshop.specialmsmqmessaging"/>
orderprocessor是一个控制台应用程序，不过可以根据需求将其设计为windows service。它的目的就是接收消息队列中的订单数据，然后将其插入到order和inventory数据库中。它利用了多线程技术，以达到提高系统性能的目的。
在orderprocessor应用程序中，主函数main用于控制线程，而核心的执行任务则由方法processorders()实现：
private static void processorders()
{
// the transaction timeout should be long enough to handle all of orders in the batch
timespan tstimeout = timespan.fromseconds(convert.todouble(transactiontimeout * batchsize));
order order = new order();
while (true)
{
// queue timeout variables
timespan datetimestarting = new timespan(datetime.now.ticks);
double elapsedtime = 0;
int processeditems = 0;
arraylist queueorders = new arraylist();
using (transactionscope ts = new transactionscope(transactionscopeoption.required, tstimeout))
{
// receive the orders from the queue
for (int j = 0; j < batchsize; j++)
{
try
{
//only receive more queued orders if there is enough time
if ((elapsedtime + queuetimeout + transactiontimeout) < tstimeout.totalseconds)
{
queueorders.add(order.receivefromqueue(queuetimeout));
}
else
{
j = batchsize; // exit loop
}
//update elapsed time
elapsedtime = new timespan(datetime.now.ticks).totalseconds - datetimestarting.totalseconds;
}
catch (timeoutexception)
{
//exit loop because no more messages are waiting
j = batchsize;
}
}
//process the queued orders
for (int k = 0; k < queueorders.count; k++)
{
order.insert((orderinfo)queueorders[k]);
processeditems++;
totalordersprocessed++;
}
//batch complete or msmq receive timed out
ts.complete();
}
console.writeline("(thread id " + thread.currentthread.managedthreadid + ") batch finished, " + processeditems + " items, in " + elapsedtime.tostring() + " seconds.");
}
}
首先，它会通过petshop.bll.order类的公共方法receivefromqueue()来获取消息队列中的订单数据，并将其放入到一个arraylist对象中，然而再调用petshop.bll.order类的insert方法将其插入到order和inventory数据库中。
在petshop.bll.order类中，并不是直接执行插入订单的操作，而是调用了iorderstrategy接口的insert()方法：
public void insert(orderinfo order)
{
// call credit card procesor
processcreditcard(order);
// insert the order (a)synchrounously based on configuration
orderinsertstrategy.insert(order);
}
在这里，运用了一个策略模式，类图如下所示：
在petshop.bll.order类中，仍然利用配置文件来动态创建iorderstategy对象：
private static readonly petshop.ibllstrategy.iorderstrategy orderinsertstrategy = loadinsertstrategy();
private static petshop.ibllstrategy.iorderstrategy loadinsertstrategy()
{
// look up which strategy to use from config file
string path = configurationmanager.appsettings["orderstrategyassembly"];
string classname = 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是一个单独的应用程序，因此它使用的配置文件与petshop不同，是存放在应用程序的app.config文件中，在该文件中，对iorderstategy的配置为：
<add key="orderstrategyassembly" value="petshop.bll" /> <add key="orderstrategyclass" value="petshop.bll.ordersynchronous" />
因此，以异步方式插入订单的流程如下图所示：
microsoft messaging queue(msmq)技术除用于异步处理以外，它主要还是一种分布式处理技术。分布式处理中，一个重要的技术要素就是有关消息的处理，而在system.messaging命名空间中，已经提供了message类，可以用于承载消息的传递，前提上消息的发送方与接收方在数据定义上应有统一的接口规范。
msmq在分布式处理的运用，在我参与的项目中已经有了实现。在为一个汽车制造商开发一个大型系统时，分销商dealer作为.net客户端，需要将数据传递到管理中心，并且该数据将被oracle的ebs(e-business system)使用。由于分销商管理系统（dms）采用的是c/s结构，数据库为sql server，而汽车制造商管理中心的ebs数据库为oracle。这里就涉及到两个系统之间数据的传递。
实现架构如下：
首先dealer的数据通过msmq传递到msmq server，此时可以将数据插入到sql server数据库中，同时利用ftp将数据传送到专门的文件服务器上。然后利用ibm的eai技术（企业应用集成，enterprise application itegration）定期将文件服务器中的文件，利用接口规范写入到eai数据库服务器中，并最终写道ebs的oracle数据库中。
上述架构是一个典型的分布式处理结构，而技术实现的核心就是msmq和eai。由于我们已经定义了统一的接口规范，在通过消息队列形成文件后，此时的数据就已经与平台无关了，使得在.net平台下的分销商管理系统能够与oracle的ebs集成起来，完成数据的处理。
以上就是petshop数据访问层消息处理部分的全部内容，希望能给大家一个参考，也希望大家多多支持。