虽然activemq+jencks的jms轻量级解决方案已经很好地在psa中work了,尤其spring的JmsTemplate使得代码更简单,但是还是存在问题。
问题来自暑期做psa的时候,linke突然提出要求,需要MDP返回些处理信息,比如处理结果、异常,以便后台监控和前台显示,但是,MDP没有返回值也没法返回异常,当时我只能无奈。
Lingo解决了这个问题,它扩展了JMS,允许异步的函数调用。
在下载了lingo-1.0-M1后(虽然1.2.1发布了,但是还没有文档支持,所以暂且用1.0),参考其自带的example,了解了它异步函数调用的代码思路。
在下载了lingo-1.0-M1后(虽然1.2.1发布了,但是还没有文档支持,所以暂且用1.0),参考其自带的example,了解了它异步函数调用的代码思路。
客户端拥有服务端的方法接口,客户端将callback和相关参数代入接口,进行异步调用,而服务端的接口实现中利用callback来返回必要的信息。
callback实现了EventListener,提供了返回值和异常的接口,另外涉及到两个方面,首先,callback本身需要轮询,其次,callback可以由实例池管理。
callback实现了EventListener,提供了返回值和异常的接口,另外涉及到两个方面,首先,callback本身需要轮询,其次,callback可以由实例池管理。
第一个方面主要参考了lingo的example,使用semaphore来进行轮询。
第二个方面并没有利用实例池,而是利用ThreadPoolExecutor来newFixedThreadPool,管理不同的异步调用线程,来完成对callback的调度。
第二个方面并没有利用实例池,而是利用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>
<bean id="broker"
class="org.activemq.spring.BrokerFactoryBean">
< property name="config" value="classpath:activemq.xml" />
< /bean>
< property name="config" value="classpath:activemq.xml" />
< /bean>
<bean id="jmsFactory"
class="org.activemq.ActiveMQConnectionFactory">
< property name="brokerURL">
< value>tcp://localhost:61616</value>
< /property>
< /bean>
class="org.activemq.ActiveMQConnectionFactory">
< 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 index="0">
< value>lingo.demo</value>
< /constructor-arg>
< /bean>
<bean
id="invocationFactory"
class="org.logicblaze.lingo.LingoRemoteInvocationFactory">
< constructor-arg>
< bean class="org.logicblaze.lingo.SimpleMetadataStrategy">
< !-- 允许单向异步调用 -->
< constructor-arg value="true" />
< /bean>
< /constructor-arg>
< /bean>
class="org.logicblaze.lingo.LingoRemoteInvocationFactory">
< 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="serviceInterface"
value="org.openepo.jms.lingo.MailService" />
< property name="connectionFactory" ref="jmsFactory" />
< property name="destination" ref="destination" />
< bean id="client"
class="org.logicblaze.lingo.jms.JmsProxyFactoryBean">
< property name="serviceInterface"
value="org.openepo.jms.lingo.MailService" />
< property name="connectionFactory" ref="jmsFactory" />
< property name="destination" ref="destination" />
<!-- 允许客户端单向异步调用 -->
< property name="remoteInvocationFactory"
ref="invocationFactory" />
< /bean>
< property name="remoteInvocationFactory"
ref="invocationFactory" />
< /bean>
<!-- 服务端配置 -->
< bean id="server"
class="org.logicblaze.lingo.jms.JmsServiceExporter">
< property name="service" ref="serverImpl" />
< property name="serviceInterface"
value="org.openepo.jms.lingo.MailService" />
< property name="connectionFactory" ref="jmsFactory" />
< property name="destination" ref="destination" />
< property name="invocationFactory" ref="invocationFactory" />
< /bean>
< bean id="server"
class="org.logicblaze.lingo.jms.JmsServiceExporter">
< property name="service" ref="serverImpl" />
< property name="serviceInterface"
value="org.openepo.jms.lingo.MailService" />
< property name="connectionFactory" ref="jmsFactory" />
< property name="destination" ref="destination" />
< property name="invocationFactory" ref="invocationFactory" />
< /bean>
<!-- 服务端代码实现 -->
< bean id="serverImpl" class="org.openepo.jms.lingo.MailServiceImpl" />
< bean id="serverImpl" class="org.openepo.jms.lingo.MailServiceImpl" />
<!-- 管理callback池,处理回调结果 -->
< bean id="asyncManager" class="org.openepo.jms.lingo.AsyncManager" singleton="false">
< property name="mailClient" ref="client" />
< property name="threadSize" value="5" />
< /bean>
< 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);
public void onResult(Object result);
// lifecycle end methods
public void stop();
public void onException(Exception e);
}
public void stop();
public void onException(Exception e);
}
ResultListenerImpl.java:
package org.openepo.jms.lingo;
import java.util.ArrayList;
import java.util.List;
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;
}
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;
}
onException = e;
}
public Exception getOnException() {
return onException;
}
return onException;
}
public List getResults() {
return results;
}
return results;
}
public boolean isStopped() {
return stopped;
}
return stopped;
}
public void waitForAsyncResponses(int messageCount)
{
System.out.println("Waiting for: " + messageCount + " responses to arrive");
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;
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();
}
return results.isEmpty();
}
protected synchronized boolean hasReceivedResponses(int
messageCount) {
return results.size()> = messageCount;
}
return results.size()> = messageCount;
}
public long getWaitTime() {
return waitTime;
}
return waitTime;
}
public void setWaitTime(long waitTime) {
this.waitTime = 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);
}
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.");
}
try {
for (Mail mail : mails) {
sendMail(mail);
Thread.sleep(2000);// 服务端时耗
listener.onResult(mail.getContent() + " Sended Successfully.");
}
listener.stop();
} catch (Exception e) {
listener.onException(e);
}
}
} catch (Exception e) {
listener.onException(e);
}
}
public void sendMail(Mail mail) throws Exception {
// 可以取消下面的注释来查看服务端将异常传给客户端
//throw new Exception("Error occurs on server side.");
}
}
// 可以取消下面的注释来查看服务端将异常传给客户端
//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;
}
private static final long serialVersionUID = 1L;
private String content;
public String getContent() {
return content;
}
public void setContent(String content) {
this.content = content;
}
this.content = content;
}
public Mail(String content) {
this.content = 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;
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池
.newFixedThreadPool(threadSize); //callback池
public void setThreadSize(int threadSize) {
AsyncManager.threadSize = threadSize;
}
AsyncManager.threadSize = threadSize;
}
private MailService mailClient;
public void setMailClient(MailService mailClient) {
this.mailClient = 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);
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());
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中的信息,进行异常处理等。
// 服务端异常
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"));
@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);
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();
}
}
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的代码量。