使用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>
 <bean id="broker" class="org.activemq.spring.BrokerFactoryBean">
  < 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>
 <bean id="destination" class="org.activemq.message.ActiveMQQueue">
  < 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>
 <!-- 客户端配置 -->
 < 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>
 <!-- 服务端配置 -->
 < 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" />
 <!-- 管理callback池,处理回调结果 -->
 < 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的代码量。

使用Lingo增强JMS

上一篇:关于CGContext画图的问题


下一篇:Splay 伸展树