JBOSS建立JMS应用实例
环境配置说明
Jboss4.2.3 Java ee 5 MySql5.0
安装jdk,我的目录为C:\\Sun\\SDK 安装jboss.我的目录为D:\\jboss4.2.3 安装mysql
配置环境变量JBOSS_HOME D:\\jboss4.2.3 配置环境变量 JAVA_HOME C:\\Sun\\SDK\\jdk
JBOSS的JMS配置
由于没有用到集群所以用JBOSS的default应用即可。 1、 配置MySQL数据库的JNDI
将MySQL数据库驱动拷到default\\lib下JBOSS没有自带MySQL的数据库驱动。 在D:\\jboss4.2.3\\docs\\examples\\jca 下的文件夹下面，有很多不同数据库引用的数据源定义模板。将其中的 mysql-ds.xml 文件Copy到你使用的服务器下，如 D:\\ jboss4.2.3\\server\\default\\deploy。 修改 mysql-ds.xml 文件的内容，使之能通过JDBC正确访问你的MySQL数据库，如下：
Sorter
这里，定义了一个名为MySqlDS的数据源，其参数包括JDBC的URL，驱动类名，用户名及密码等。
通过配置后就可以通过JNDI名java:MySqlDS 访问到配置的Mysql数据库。
2、 配置JBOSS的JMS环境
将D:\\jboss4.2.3\\docs\\examples\\jms下的mysql-jdbc2-service.xml拷到D:\\jboss4.2.3\\server\\default\\deploy\\jms 下。并将数据库DataSourceBinding 改成name=MySqlDS连到你的MySql数据库
将D:\\jboss4.2.3\\server\\default\\deploy\\jms 目录下的hsqldb-jdbc-state-service文件改名为mysql-jdbc-state-service.xml 并将
该成name=MySqlDS用于连接你的MySql的数据库。
启动JBOSS后在控制台将会看到类似如下的信息就表示默认的JMS的jndi名已经绑定了。JMS的配置成功了。
10:14:42,343 INFO [ConnectionFactoryBindingService] Bound ConnectionManager
'jboss.jca:service=DataSourceBinding,name=DefaultDS' to JNDI name 'java:DefaultDS'
10:14:42,984 INFO [ConnectionFactoryBindingService] Bound ConnectionManager
'jboss.jca:service=ConnectionFactoryBinding,name=JmsXA' to JNDI name 'java:JmsXA'
10:14:43,171 INFO [ConnectionFactoryBindingService] Bound
ConnectionManager 'jboss.jca:service=DataSourceBinding,name=MySqlDS' to JNDI name 'java:MySqlDS'
10:14:43,968 INFO [A] Bound to JNDI name: queue/A 10:14:44,000 INFO [B] Bound to JNDI name: queue/B 10:14:44,000 INFO [C] Bound to JNDI name: queue/C 10:14:44,000 INFO [D] Bound to JNDI name: queue/D 10:14:44,015 INFO [ex] Bound to JNDI name: queue/ex
10:14:44,046 INFO [testTopic] Bound to JNDI name: topic/testTopic 10:14:44,046 INFO [securedTopic] Bound to JNDI name: topic/securedTopic 10:14:44,062 INFO [testDurableTopic] Bound to JNDI name:
topic/testDurableTopic
10:14:44,062 INFO [testQueue] Bound to JNDI name: queue/testQueue 10:14:44,140 INFO [myQueue] Bound to JNDI name: queue/myQueue
10:14:44,203 INFO [UILServerILService] JBossMQ UIL service available at : /127.0.0.1:8093
连接到MySQL数据库也可以看到默认的建了几个以jms_开头的数据表。用语保存jms的用户信息和持久化消息。
JBOSS的JMS实例
JBOSS的JMS环境建立好了以后我们就可以写几个发送接受JMS的程序来验证一下。 当你发送一个消息，你不能直接发送到对此消息感兴趣的接受者。而是你发送到一个目的地。对此消息感兴趣的接受者必须连接到目的地，得到此消息或在目的地设置订阅。
在JMS中有两种域：topics 和queues 。
*一个消息发送到一个topics ，可以有多个客户端。用topic发布允许一对多，或多对多通讯通道。消息的产生者被叫做publisher, 消息接受者叫做subscriber。
*queue 是另外一种方式，仅仅允许一个消息传送给一个客户。一个发送者将消息放在消息队列中，接受者从队列中抽取并得到消息，消息就会在队列中消失。第一个接受者抽取并得到消息后，其他人就不能在得到它。
为了能发送和接收消息，必须得到一个JMS连接。该连接是使用JMS Provider得到连接的，在得到连接之后，建立一个会话(Session)。然后再建立publisher/sender 来发送消息或subscriber/receiver来接收消息。
运行时，如果使用topic 那么publisher 或subscriber 通过一个topic来关联，如果使用queue ，则sender 或receiver通过queue来关联起来。
通常，在JMS框架中运转的方法如下： (1) 得到一个JNDI初始化上下文(Context)；
(2) 根据上下文来查找一个连接工厂TopicConnectFactory/ QueueConnectionFactory (有两种连接工厂，根据是topic/queue来使用相应的类型)；
(3) 从连接工厂得到一个连接(Connect 有两种[TopicConnection/ QueueConnection]); (4) 通过连接来建立一个会话(Session); (5) 查找目的地(Topic/ Queue);
(6) 根据会话以及目的地来建立消息制造者(TopicPublisher/QueueSender)和消费者(TopicSubscriber/ QueueReceiver).
为了得到一个连接和得到一个目的地（用来关联publisher/sender 或subscriber/receiver），必须用provider-specific参数。
在D:\\jboss4.2.3\\server\\default\\deploy\\jms下的jbossmq-destinations-service.xml是在配置JMS目的地的xml文件。在文件中已经存在几个缺省的目的地，所以你比较容易明白怎样增加到文件中
1、 topics例子 package com.msg;
import javax.naming.Context;
import javax.naming.InitialContext; import javax.naming.NamingException; import javax.jms.TopicConnectionFactory; import javax.jms.TopicConnection; import javax.jms.TopicSession; import javax.jms.TopicPublisher; import javax.jms.Topic;
import javax.jms.TextMessage; import javax.jms.Session;
import javax.jms.JMSException; import java.util.Hashtable;
public class HelloPublisher { TopicConnection topicConnection; TopicSession topicSession; TopicPublisher topicPublisher; Topic topic; public HelloPublisher(String factoryJNDI, String topicJNDI) throws JMSException, NamingException { Hashtable props = new Hashtable(); props.put(Context.INITIAL_CONTEXT_FACTORY, \ props.put(Context.PROVIDER_URL, \ props.put(\ props.put(Context.URL_PKG_PREFIXES, \ Context context = new InitialContext(props); TopicConnectionFactory topicFactory = (TopicConnectionFactory) context .lookup(factoryJNDI); topicConnection = topicFactory.createTopicConnection(); topicSession = topicConnection.createTopicSession(false,
}
}
Session.AUTO_ACKNOWLEDGE);
topic = (Topic) context.lookup(topicJNDI); topicPublisher = topicSession.createPublisher(topic);
public void publish(String msg) throws JMSException { }
TextMessage message = topicSession.createTextMessage(); message.setText(msg);
topicPublisher.publish(topic, message);
public void close() throws JMSException { topicSession.close(); topicConnection.close(); }
public static void main(String[] args) { try { HelloPublisher publisher = new HelloPublisher(\ \ for (int i = 1; i < 11; i++) { String msg = \ System.out.println(\ publisher.publish(msg); } publisher.close(); } catch (Exception ex) { System.err .println(\ + ex); ex.printStackTrace(); } }
package com.msg;
import javax.naming.Context;
import javax.naming.InitialContext; import javax.naming.NamingException;
import javax.jms.TopicConnectionFactory; import javax.jms.TopicConnection; import javax.jms.TopicSession; import javax.jms.TopicSubscriber; import javax.jms.Topic; import javax.jms.Message; import javax.jms.TextMessage; import javax.jms.Session;
import javax.jms.MessageListener; import javax.jms.JMSException;
public class HelloSubscriber implements MessageListener { TopicConnection topicConnection; TopicSession topicSession; TopicSubscriber topicSubscriber; Topic topic; public HelloSubscriber(String factoryJNDI, String topicJNDI) throws JMSException, NamingException { Context context = new InitialContext(); TopicConnectionFactory topicFactory = (TopicConnectionFactory) context .lookup(factoryJNDI); topicConnection = topicFactory.createTopicConnection(); topicSession = topicConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); topic = (Topic) context.lookup(topicJNDI); topicSubscriber = topicSession.createSubscriber(topic); topicSubscriber.setMessageListener(this); System.out.println(\ topicConnection.start(); } public void onMessage(Message m) { try { String msg = ((TextMessage) m).getText(); System.out.println(\ } catch (JMSException ex) { System.err.println(\ ex.printStackTrace(); } } public void close() throws JMSException {
topicSession.close(); topicConnection.close(); } public static void main(String[] args) { try { HelloSubscriber subscriber = new HelloSubscriber( \ } catch (Exception ex) { System.err .println(\ + ex); ex.printStackTrace(); } } }
2、 queues 例子 package com.msg;
import javax.naming.Context;
import javax.naming.InitialContext; import javax.naming.NamingException; import javax.jms.QueueConnectionFactory; import javax.jms.QueueConnection; import javax.jms.QueueSession; import javax.jms.QueueSender; import javax.jms.Queue;
import javax.jms.TextMessage; import javax.jms.Session;
import javax.jms.JMSException;
import java.util.Hashtable;
public class HelloQueue {
QueueConnection queueConnection; QueueSession queueSession; QueueSender queueSender; Queue queue;
public HelloQueue(String factoryJNDI, String topicJNDI) throws JMSException, NamingException { Hashtable props = new Hashtable(); props.put(Context.INITIAL_CONTEXT_FACTORY,
}
\props.put(Context.PROVIDER_URL, \props.put(\
props.put(Context.URL_PKG_PREFIXES, \Context context = new InitialContext(props);
QueueConnectionFactory queueFactory = (QueueConnectionFactory) context .lookup(factoryJNDI);
queueConnection = queueFactory.createQueueConnection(); queueSession = queueConnection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE); queue = (Queue) context.lookup(topicJNDI); queueSender = queueSession.createSender(queue);
public void send(String msg) throws JMSException { TextMessage message = queueSession.createTextMessage(); message.setText(msg); queueSender.send(queue, message); }
public void close() throws JMSException { queueSession.close(); queueConnection.close(); }
public static void main(String[] args) { try { HelloPublisher publisher = new HelloPublisher(\ \ for (int i = 1; i < 11; i++) { String msg = \ System.out.println(\ publisher.publish(msg); } publisher.close(); } catch (Exception ex) { System.err .println(\ + ex); ex.printStackTrace(); }
} }
package com.msg;
import java.util.Hashtable;
import javax.jms.JMSException; import javax.jms.Message;
import javax.jms.MessageListener; import javax.jms.Session;
import javax.jms.TextMessage; import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory; import javax.jms.QueueSession; import javax.jms.QueueReceiver; import javax.naming.Context;
import javax.naming.InitialContext; import javax.naming.NamingException;
public class HelloReceQueue implements MessageListener { QueueConnection queueConnection; QueueSession queueSession; QueueReceiver queueReceiver; Queue queue;
public HelloReceQueue(String factoryJNDI, String topicJNDI) throws JMSException, NamingException { Context context = new InitialContext(); QueueConnectionFactory queueFactory = (QueueConnectionFactory)context.lookup(factoryJNDI); queueConnection = queueFactory.createQueueConnection(); queueSession = queueConnection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE); queue = (Queue)context.lookup(topicJNDI); queueReceiver = queueSession.createReceiver(queue); queueReceiver.setMessageListener(this); System.out.println( \ queueConnection.start(); }
public void onMessage(Message m) { try {
String msg = ((TextMessage)m).getText();
System.out.println(\ } catch(JMSException ex) {
System.err.println(\ ex.printStackTrace(); } }
public void close() throws JMSException { queueSession.close(); queueConnection.close(); } }
我们通过一个jsp页面建立一个web应用来发送接受消息观察message的发送接受情况。 sendmsg.jsp
<%@ page language=\<%@ page import=\<%
String path = request.getContextPath(); String basePath =
request.getScheme()+\HelloPublisher sendmsgBean=new HelloPublisher(\
String msg=request.getParameter(\
String flag=request.getParameter(\
if(!\{
sendmsgBean.publish(msg); }
if(\{
out.println(\
HelloSubscriber subscriber=new HelloSubscriber(\}
if(\{
HelloQueue queueBean=new HelloQueue(\ queueBean.send(msg); }
if(\{
HelloReceQueue receQueueBean=new HelloReceQueue(\\
}
out.println(flag); %>