使用Lingo增强JMS

虽然activemq+jencks的jms轻量级解决方案已经很好地在psa中work了，尤其spring的JmsTemplate使得代码更简单，但是还是存在问题。

问题来自暑期做psa的时候，linke突然提出要求，需要MDP返回些处理信息，比如处理结果、异常，以便后台监控和前台显示，但是，MDP没有返回值也没法返回异常，当时我只能无奈。

Lingo解决了这个问题，它扩展了JMS，允许异步的函数调用。
在下载了lingo-1.0-M1后(虽然1.2.1发布了，但是还没有文档支持，所以暂且用1.0)，参考其自带的example，了解了它异步函数调用的代码思路。

客户端拥有服务端的方法接口，客户端将callback和相关参数代入接口，进行异步调用，而服务端的接口实现中利用callback来返回必要的信息。
callback实现了EventListener，提供了返回值和异常的接口，另外涉及到两个方面，首先，callback本身需要轮询，其次，callback可以由实例池管理。

第一个方面主要参考了lingo的example，使用semaphore来进行轮询。
第二个方面并没有利用实例池，而是利用ThreadPoolExecutor来newFixedThreadPool，管理不同的异步调用线程，来完成对callback的调度。

配置部分：

<?xml version="1.0"
encoding="UTF-8"?>

<!DOCTYPE beans PUBLIC "-//SPRING//DTD
BEAN//EN" "http://www.springframework.org/dtd/spring-beans.dtd">

<beans>

<property
name="brokerURL">
<value>tcp://localhost:61616</value>

</property>
</bean>

<bean id="destination"
class="org.activemq.message.ActiveMQQueue">
<constructor-arg
index="0">
<value>lingo.demo</value>

</constructor-arg>
</bean>

<constructor-arg>
<bean
class="org.logicblaze.lingo.SimpleMetadataStrategy">

<constructor-arg
value="true" />
   </bean>
  </constructor-arg>
</bean>

<bean id="client"
class="org.logicblaze.lingo.jms.JmsProxyFactoryBean">

<property
name="remoteInvocationFactory"
   ref="invocationFactory" />
</bean>

<bean id="server"
class="org.logicblaze.lingo.jms.JmsServiceExporter">

<bean id="serverImpl"
class="org.openepo.jms.lingo.MailServiceImpl"
/>

<bean id="asyncManager"
class="org.openepo.jms.lingo.AsyncManager"
singleton="false">
<property name="mailClient" ref="client"
/>
<property name="threadSize" value="5"
/>
</bean>

</beans>

ResultListener和ResultListenerImpl：callback接口及实现。

ResultListener.java：

package org.openepo.jms.lingo;

import java.util.EventListener;

public interface ResultListener extends EventListener {
public void onResult(Object result);

    // lifecycle end methods
    public void stop();
    public void onException(Exception e);
}

ResultListenerImpl.java：

package org.openepo.jms.lingo;

import java.util.ArrayList;
import java.util.List;

public class ResultListenerImpl implements ResultListener
{
    private List results = new ArrayList();
    private Object semaphore = new Object();
    private boolean stopped;
    private Exception onException;
    private long waitTime = 1000;

    public synchronized void onResult(Object result)
{

results.add(result);
        synchronized (semaphore) {

semaphore.notifyAll();

}

}

    // lifecycle end methods
    public void stop() {
        stopped = true;
    }

    public void onException(Exception e)
{
        onException = e;
    }

    public Exception getOnException()
{
        return
onException;
    }

    public List getResults()
{
        return results;
    }

    public boolean isStopped()
{
        return stopped;
    }

public void waitForAsyncResponses(int
messageCount) {

System.out.println("Waiting for: " + messageCount +
" responses to arrive");

long start =
System.currentTimeMillis();

        for (int i = 0; i
< 10; i++) {

try {

if (hasReceivedResponses(messageCount))
{

break;

}

synchronized (semaphore) {

semaphore.wait(waitTime);

}

}

catch (InterruptedException e) {

System.out.println("Caught: " + e);

}
        }
        long end =
System.currentTimeMillis() - start;

System.out.println("End of wait for " + end + " millis");
}

    protected boolean hasReceivedResponse()
{
        return
results.isEmpty();
    }

    protected synchronized boolean
hasReceivedResponses(int messageCount) {
        return results.size()
>= messageCount;
    }

    public long getWaitTime()
{
        return waitTime;
    }

    public void setWaitTime(long waitTime)
{
        this.waitTime =
waitTime;
    }
}

MailService和MailServiceImpl：服务代码。

MailService.java：

package org.openepo.jms.lingo;

import java.util.List;

public interface MailService {
public void
asyncSendMail(List<Mail> mails,
ResultListener listener);
}

MailServiceImpl.java：

package org.openepo.jms.lingo;

import java.util.List;

public class MailServiceImpl implements MailService
{

    public void
asyncSendMail(List<Mail> mails,
ResultListener listener) {

      try {

for (Mail mail : mails) {

sendMail(mail);

Thread.sleep(2000);// 服务端时耗

listener.onResult(mail.getContent() +
" Sended Successfully.");

}

listener.stop();
        } catch
(Exception e) {

listener.onException(e);
        }
    }

    public void sendMail(Mail mail) throws
Exception {
      // 可以取消下面的注释来查看服务端将异常传给客户端
      //throw new Exception("Error occurs
on server side.");
    }
}

在服务端方法中,可以利用callback将处理结果,是否结束和异常信息返回客户端.

Mail.java：

package org.openepo.jms.lingo;

import java.io.Serializable;

public class Mail implements Serializable
{

private static final long serialVersionUID = 1L;

private String content;

public String getContent() {
return content;
}

public void setContent(String content)
{
this.content = content;
}

public Mail(String content) {
this.content = content;
}
}

AsyncManager：各类异步调用的方法可以集中在这个类中，利用线程池来统一控制callback实例。

AsyncManager.java：

package org.openepo.jms.lingo;

import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;

public class AsyncManager {

static private int threadSize =
10; //callback池大小

static private ThreadPoolExecutor executor =
(ThreadPoolExecutor) Executors
.newFixedThreadPool(threadSize); //callback池

public void setThreadSize(int threadSize)
{
AsyncManager.threadSize = threadSize;
}

private MailService mailClient;

public void setMailClient(MailService mailClient)
{
this.mailClient = mailClient;
}

public AsyncManager() {
}

public void sendMails(final
List<Mail> mails)
{

  // callback对象
  final ResultListenerImpl callBack = new
ResultListenerImpl();
  callBack.setWaitTime(2000);

  // 异步调用
  mailClient.asyncSendMail(mails, callBack);

  // 调用线程池中的callback
  executor.execute(new Runnable()
{
   public void run() {
    // callBack 阻塞等待n个消息
    callBack.waitForAsyncResponses(mails.size());

    if (callBack.getOnException() != null) {
     // 服务端异常
     System.out.println("Server
Exception: "
       +
callBack.getOnException().getMessage());
    } else
{
     // 得到服务端处理结果,打印结果
     for (Object result :
callBack.getResults())
{
      System.out.println("Result:
" + result);
     }
    }
   }
  });
}
}
上面匿名类的run方法中，在callback的waitForAsyncResponses方法结束后，可以检查callback中的信息，进行异常处理等。

下面是测试用例：
@Test
public void test() {
List<Mail> mails = new
ArrayList<Mail>();
mails.add(new Mail("mail1"));
mails.add(new Mail("mail2"));

// 计算时间
long startTime = System.currentTimeMillis();

try {
  // 异步调用
  asyncManager.sendMails(mails);

  // 没有阻塞
  System.out.println("Cost time "
     + (System.currentTimeMillis() -
startTime) + "ms");

  //为查看结果,sleep主线程
  Thread.sleep(10000);
} catch (InterruptedException e)
{
  e.printStackTrace();
}
}

使用Lingo对JMS增强后，通过callback，使得异步调用的确比较OO了，但是更重要的是服务端的信息可以通过callback返回给客户端，客户端可以相应地进行处理。

多出了许多代码，自然复杂度有所增加，但是lingo-1.2.1后，增加了annotation，减少了callback的代码量。