用JMS客户端利用空闲的计算机资源

　　摘要
　　在企业中，许多计算机由于在其上执行工作的性质而未得到充分利用，或者因为过了上班时间而干脆得不到使用。在许多情况下，应用服务器耗光了宝贵的CPU（尤其是在执行CPU密集型的数学运算时），而网络上的其他计算机则闲置一旁。本文提出了一种框架，用于把Java消息服务（Java Messaging Service，JMS）客户端放置在这些未充分利用的计算机上，以分担一些通常应由服务器执行的工作。该客户端可以监听某个要执行的工作单元的请求队列，然后在应答队列中做出响应。此外，本文还给出了一个BEA WebLogic Integration 8.1架构的例子，它通过把一个工作流以及相关的Java控件用作替代框架，把工作分发到远程客户端上，从而把工作单元可靠地分发给JMS请求队列。

　　简介
　　本文提出了一种J2EE框架，用于解决把工作分配给未充分利用的计算机资源这个难题。具体来说，可以把JMS客户端放置在这些未充分利用的计算机上，从而分担一些通常应由服务器执行的工作。该客户端可以监听某个要执行的工作单元的请求队列，然后在应答队列上做出响应。可以使用一组消息驱动bean获取应答队列上的响应消息，以便进行进一步的处理。此外，还可以使用一种servlet实现来管理性地启动用于创建（要发送给JMS客户端的）工作单元的整个子流程，并使用它来终止这个子流程。

　　我使用常见的BEA WebLogic Server作为把离散的工作单元分配给分布式JMS客户端的例子。在另一个更为复杂的例子中，BEA WebLogic Integration (WLI)工作流也执行类似的分发任务，但是通过对请求队列进行监控，它在灵活性、Java控件的可重用性和可伸缩性等方面要更好一些。

　　使用的例子
　　业内有相当多的例子可以演示如何使用JMS框架来充分利用计算机进行并行处理。例如：

　　一个银行应用程序可以实现抵押贷款，并以不同的比率和年份执行几类利息计算，从而为信贷官员提供与每个申请者相关的、可能影响借贷类型的数据。所有不同种类的计算可以按照申请者分配给可用的计算机来执行，然后把结果返回给应用服务器储存起来。
　　一个记帐系统可以从数据库读取记录，然后重新计算记录中的数字，以求做到更加精确。对于每条记录，它可能需要连接到业务用户的本地系统中以获得辅助数据，这可能需要几秒钟。如果顺序执行，当涉及到的记录上千时，这种方法不仅很慢，而且可能进一步延长服务器线程等待从各地返回响应的时间。通过把这些工作分发给JMS客户端，不仅可以并行完成处理，而且还可以节省服务器线程。
　　一个天气预报系统或线性优化系统可能需要操纵或执行矩阵乘法。随着矩阵的大小和数量逐步增加，服务器CPU的负担也随之加重。如果这种情况经常发生，那么通过把矩阵操作和乘法分发给其他计算机上的JMS客户端，服务器的CPU就可以节省下来用于其他工作。
使用常规的WebLogic Server来分发工作单元
　　借鉴最后一个例子，我将构建一个简单的例子，用于执行矩阵乘法，同时说明如何使用JMS框架把计算工作分发给企业中的计算机资源。JMS客户端将收到一个工作单元实例，之后，它将会调用其doWork()方法。在这个简单的例子中，doWork()方法将把2个3×3的矩阵相乘，然后把结果保存到一个结果矩阵中。
接着，JMS客户端使用工作单元实例的一个副本（工作是在这上面执行的）对应答队列做出响应。一个消息驱动bean将接受已完成的工作。图1说明了我将要讨论的各个组件：

图 1.该WebLogic Server实现中的各个组件

　　这种方法以常规方式使用了JMS系统。在下面的内容中，我将引入一些代码，并考虑几个扩展问题。

　　工作单元类
　　JMS请求队列上的每个类都将实现一个UnitOfWork接口，该接口有一个特别有趣的方法，叫做doWork()：

public interface UnitOfWork extends java.io.Serializable {
  // This method executes itself on the client machine
  public void doWork();
  // This method prints the current contents of performed work
  public void print();
  // This method stores the instance into a backing store
  public void store();
}

　　在我们这个简洁而直观的例子中，我使用一个SimpleMatri类实现了UnitOfWork接口：

public class SimpleMatrix implements UnitOfWork {
 private Integer m1[][];
 private Integer m2[][];
 private Integer result[][];
 private Integer rows = new Integer(3);
 private Integer cols = new Integer(3);
 // May initialize m1 and m2 by locating records from a database
 public SimpleMatrix() {
 }
 // This method actually multiplies m1 x m2 and stores in result
 public void doWork() {
 }
 // This method stores result into a backing store
 public void store() {
 }
 // This method prints the current contents of result
 public void print() {
 }
}

　　方法的实现相当简单，限于文章的篇幅，这里就不再进行说明。请参见所附的示例代码，其中给出了完整实现。这里的要点在于，这个SimpleMatrix实例被传递给一个JMS客户端，该客户端只要调用doWork()即可利用其CPU来执行工作。对于这个例子，我不会实际从数据库中检索矩阵或者把矩阵保存到数据库中，但是在实际应用中，这是必须完成的工作。

　　Servlet工作创建程序
　　可以使用一个servlet来创建这些UnitOfWork实例。尽管WebLogic Server启动类可以执行同样的功能，但出于管理的目的，从安全的Web浏览器发送消息给servlet要更加容易。（另一种可选的实现是使用Web服务。）如果在servlet上安置了安全性，通过身份验证的用户可以在查询字符串中传递命令，以便开始交付工作单元给JMS请求队列或停止交付。我将给出一个servlet的主干例子，以说明其中的一些有用方法。

public class WorkServlet extends HttpServlet {
 ...
 private QueueSender qsender;
 private ObjectMessage msg;
 private int numMessages = 5;
 ...
 // This places the unit of work on the request queue
 public synchronized boolean sendMessages(int numberOfMessages,
                                                 PrintWriter o) {
   for(int i=0; i
　　JMS客户端类的任务仅仅是接受请求队列上的消息，调用对象上的doWork()方法在这台计算机上执行工作，然后把结果返回给应答队列，消息驱动bean从应答队列中获取结果，以便进行进一步的处理和保存。可以检查它是否是文本消息，然后告诉客户端停止处理，从而允许发送控件消息给客户端。当然，在实际应用中，消息可能包含客户端的名称，这样就不会造成所有的客户端都停止处理。
　　使用UnitOfWork接口的优点在于，JMS客户端只要编写一次，就可以用于以后实现该接口的任何类。这使得JMS客户端具有很大的通用性，可以不加修改地应用到许多不同的场景中。只需把编译后的UnitOfWork接口以及它所有的实现类都包含在客户端的类路径中。
　　在这个简化模型中，客户端需要等待消息到达以启动处理。在实际情况中，客户端的方法等待的条件可以是一天中的某个时刻，比如下午5点，或者是计算机的CPU负载低于某个阈值。需要把此类逻辑添加到客户端，使之与计算机的使用安排更加一致。
消息驱动bean接收程序
　　消息驱动bean实例将监听应答队列，看看有没有已完成的工作对象单元。下面给出一个例子的主干部分：public class MessageWorkBean implements MessageDrivenBean,
                                            MessageListener {
  ...
  // This method will receive a unit of work object to store
 public void onMessage(Message msg) {
   ObjectMessage om = (ObjectMessage) msg;
   try {
     UnitOfWork unit = (UnitOfWork)om.getObject();
     unit.print();
     unit.store();
   }
   catch(JMSException ex) {
    log("Message Driven Bean: Could not retrieve Unit of Work.");
    ex.printStackTrace();
   }
 }
}

[1] [2] [3] 下一页