虽然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的调度。
配置部分：
encoding="UTF-8"?>
/p>
BEAN//EN" "http://www.springframework.org/dtd/spring-beans.dtd">
class="org.activemq.spring.BrokerFactoryBean">
value="classpath:activemq.xml"
/>
class="org.activemq.ActiveMQConnectionFactory">
name="brokerURL">
tcp://localhost:61616
class="org.activemq.message.ActiveMQQueue">
index="0">
lingo.demo
id="invocationFactory"
class="org.logicblaze.lingo.LingoRemoteInvocationFactory">
class="org.logicblaze.lingo.SimpleMetadataStrategy">
value="true" />
class="org.logicblaze.lingo.jms.JmsProxyFactoryBean">
value="org.openepo.jms.lingo.MailService"
/>
ref="jmsFactory" />
ref="destination" />
name="remoteInvocationFactory"
ref="invocationFactory" />
class="org.logicblaze.lingo.jms.JmsServiceExporter">
ref="serverImpl" />
value="org.openepo.jms.lingo.MailService"
/>
ref="jmsFactory" />
ref="destination" />
ref="invocationFactory" />
class="org.openepo.jms.lingo.MailServiceImpl"
/>
class="org.openepo.jms.lingo.AsyncManager"
singleton="false">
/>
/>
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 mails,
ResultListener listener);
}
MailServiceImpl.java：
package org.openepo.jms.lingo;
import java.util.List;
public class MailServiceImpl implements MailService
{
public void
asyncSendMail(List 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 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 mails = new
ArrayList();
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的代码量。
