近日开发一个Message Driven Bean实现的异步计算功能,整理之后希望可以和各位正在做MDB或者JMS开发的朋友分享一下。
这是一个MIS系统的一部分,用于计算最后的收费金额,由于计算过程及其复杂,要求实现不影响用户接口响应,就是要实现异步计算,同时可以处理多个计算请求,计算结束后通知发出计算命令的客户端。功能十分简单,只要一个MDB就可以实现了,但是要通知客户端计算结束使得系统变得有趣多了。
JMS API
在进行系统设计之前我们先回头熟悉一下JMS的一些概念。JMS源于企业应用对于消息中间件的需求,使应用程序可以通过消息进行异步处理而互不影响。JMS应用程序有四个组成部分:JMS服务提供者、消息管理对象、消息的生产者消费者和消息本身。
l JMS服务提供者实现消息队列和通知,同时实现消息管理的API。JMS已经是J2EE API的一部分,J2EE服务器都提供JMS服务。
l 消息管理对象提供对消息进行操作的API。JMS API中有两个消息管理对象:ConnectionFactory和Destination,根据消息的消费方式的不同ConnectionFactory可以分为QueueConnectionFactory和TopicConnectionFactory,Destination可以分为Queue和Topic。用这两个管理对象可以建立到消息服务的会话。
l 消息的生产者和消费者。它们可以毫不相干,只需要消息的消费者知道如何使用消息即可。根据消息消费者数目的不同,消息的消费者分为两类:subscriber 和receiver,同样消息发送者也分为两类:Publisher和Sender。
l 消息。JMS API规定了五种消息:Message、MapMessage、TextMessage、ByteMessage、StreamMessage和ObjectMessage
消费形式的不同造成JMS有两组平行的API,这就是JMS的PTP(point to point)模型和PUB/SUB(publisher和subscriber、出版和订阅)模型。PTP的消息应用中一个消息只有一个消费者,消费后该消息即不再有效。而PUB/SUB应用中一个消息可以有多个订阅者,而且每个订阅者不一定非要处理该消息。
下面是JMS应用的基本模型:
(该图引自sun的在线文档 JMS Tutorial)
根据该模型列出JMS中几个最长用的API:
l QueueConnectionFactory和TopicConnectionFactory 连接工厂用来生成QueueConnection和TopicConnection的实例
l QueueConnection和TopicConnection连接对象用来建立到JMS的连接并生成会话实例
l QueueSession和TopicSession会话对象用来创建消息、消息的生产者和消息的消费者(解释一下消息的生产者:它并不代表生成消息实例的对象而是指将消息发送到JMS的对象)
l QueueSender、TopicPublisher和QueueReciever、TopicSubscriber。消息的生产者和消费者,QueueSender的send方法和TopicPublisher的publish方法发送消息到Destination。QueueReciever和TopicSubscriber直接使用父接口MessageConsumer中定义的方法receive、recieveNoWait等方法来接收消息,setMessageListener方法来设置消息监听器。QueueReciever的getQueue方法得到Queue的引用,TopicSubscriber的getTopic方法得到Topic的引用。
l MessageListener,消息监听器。改接口只有一个方法onMessage(),改方法只有一个Message类型的参数,通过MessageConsumer(QueueReciever和TopicSubscriber共同的父接口)的setMessageListener方法注册后,系统在收到消息后调用改方法。
l Queue和Topic,消息Destination。主要的作用就是存储消息。
设计
系统过于简单就不设计了,这里简单交代一下软件的运行环境和执行流程,这个MIS系统是B/S模式的,Web Browser在完成添加执行任务(保存到数据库)的工作后要发一个消息给计算程序,这个计算过程就实现成MDB,计算结束后将完成的消息发给生成任务的Web Browser,各位不要以为我真的能这样做,主动式的把消息发送给Web Browser笔者还没有这份功力,这里采用了一个取巧的方法:在Web页中使用一个Applet,该Applet本身也是一个一直运行的线程,并且每隔一段时间访问接收消息的Servlet,如果有收到消息则提示用户。这个Servlet就是一个MessageConsumer了,而MDB就是MessageProducer了。而在产生执行任务的时候,保存执行任务的jsp/servlet则是MessageProducer,而运行MDB的EJB Container则成了MessageCunsumer,MDB则成为一个消息监听器。由于有两种消息笔者偷懒就使用了两个Destination(跟上面图不谋而合,哈哈),一个用来存放执任务的消息,一个用于存放任务完成的消息。具体实现当然有消息内容的设计和实现,不过对于系统结构并不影响,这里就不废话了。
开发工具是JBuilder和WebLogic。
动手实现
1. 发送和接收消息。需要使用JMS API手动编码实现发送和接收消息的有三个地方:发送消息的Servlet、接收消息的Servlet和发送计算完成消息的MessageDrivenBean。对于发送和接收消息都需要的QueueConnectionFactory、QueueConnection、Queue和QueueSession等对象则利用Servlet和MDB的生命周期方法获得和释放,在Servlet的init方法和MDB的ejbCreate方法中获得资源并创建需要的对象,在Servlet的destroy和MDB的ejbRemove方法中释放资源。
获得资源:
public void init() throws ServletException {
try{
InputStream in = this.getClass().getClass().getClassLoader().
getResourceAsStream("jndi.properties");
Properties p = new Properties();
p.load(in);
ctx = new InitialContext(p);
}catch(Exception ex){
ex.printStackTrace();
}
try{
connectionFactory = (QueueConnectionFactory) ctx.lookup(
connectionFactoryName);
queue = (Queue) ctx.lookup(queueName);
connection = (javax.jms.QueueConnection) ( (QueueConnectionFactory)
connectionFactory).
createQueueConnection();
queueSession = ( (javax.jms.QueueConnection) connection).
createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
queueSender = queueSession.createSender(queue);
}catch(Exception ex){
ex.printStackTrace();
}
}
发送消息:
public void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
String taskid = request.getParameter("taskid");
try{
String completedmsg = "任务" + taskid + "执行完成";
ObjectMessage om = queueSession.createObjectMessage(new CalculateCompleted(userid, taskid, completedmsg));
queueSender.send(om);
}catch(Exception ex){
ex.printStackTrace();
}
}
接收消息。这里对消息在Servlet中作了转存,messages是一个Hashtable对象。
public void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
boolean rflag = true;
try {
messageConsumer = (QueueReceiver) queueSession.createReceiver(queue);
connection.start();
while (rflag) {
Message m = messageConsumer.receiveNoWait();
if (m != null) {
if (m instanceof ObjectMessage) {
ObjectMessage message = (ObjectMessage) m;
MessageBody mb = (MessageBody)message.getObject();
ArrayList a = (ArrayList)messages.get(mb.getUserID());
if(a == null){
a = new ArrayList();
a.add(mb);
messages.put(mb.getUserID(), a);
}else{
a.add(mb);
}
}
else {
rflag = false;
}
}
else {
rflag = false;
}
}
connection.stop();
}
catch (JMSException e) {
e.printStackTrace();
}
}
释放资源:
public void destroy() {
try{
if(connection != null)connection.close();
}catch(Exception ex){
ex.printStackTrace();
}
}
MDB的jebCreate方法和上面init方法的内容一样,ejbRemove和上面destroy方法的内容相同。MDB的计算过程实现在onMessage()方法中,计算完成后发送一条消息,发送消息的过程上面已有,不再赘述。
2. WebLogic的JMS服务配置。这里使用最简单的JMS配置,除了名字和JNDI名其余的均使用缺省值。
a) 启动WebLogic,打开web console
b) 展开左侧的JMS节点,新建一个JMS Server。名字随意。
c) 为刚建立的JMS Server建立两个Queue。JNDI名分别为::jms/calculate和jms/completed。
d) 建立一个ConnectionFacotry。JNDI名:jms/conn_factory
3. 配置MDB。这里除了要指定MDB监听的Queue外,因为MDB需要向另外的Queue发送计算成功的消息,还需要把上面WebLogic中配置的jms/conn_factory作为资源添加到MDB的resource-ref中,把jms/completed作为环境资源添加到MDB的resource-env-resource中。
4. 部署执行。用JBuilder建立一个EAR然后部署,非常简单。如果各位想要测试只需要建立一个简单的客户端应用程序,然后将MDB的onMessage方法简单实现为消息转发就可以了。
总结笔者初学JMS,只能说说JMS最基本的应用。JMS是一组很强大的API,不仅可以在一个应用程序中实现异步通信,也常被用来在不同的应用程序间传递数据,同时JMS也支持分布式事务,达到了企业应用的要求,笔者相信随着学习和应用的不断深入,将会发现JMS更强大的功能。