使用JBOSS收发JMS

当我们清楚了以后内容后，现在我们来用JBoss实现一个例子来加深对JBoss和JMS的了解。

在上面叙述中，我们知道明确使用JMS provider有三个基本的事情要做：配置JNDI初始化上下文，连接工厂的名字和使用目的地的名字。

当编写产品的最好的事情是不受provider-specific 影响，使代码能在不同的JMS provider之间容易移植。在此这个例子没有聚焦在开发产品上，而是解释如何使用JbossMQ来工作。

1) 初始化上下文

w 配置JNDI的一个方法是通过属性文件jndi.properties。在这个文件中使用正确的值，并且把它所在的路径包含到classpath中，它比较容获得正确初始化上下文。

jndi.properties文件的内容如下：

java.naming.factory.initial=org.jnp.interfaces.NamingContextFactory

java.naming.provider.url=localhost:1099

java.naming.factory.url.pkgs=org.jboss.naming:org.jnp.interfaces

把该文件放置的路径成为你的classpath的一部分。如果你使用这种方法，在初始化上下文时，代码比较简单: Context context = new IntialContext();1

w 在某些情景下，可能需要手工配置JNDI；例如当运行的类文件中环境已经配置了一个初始化上下文，但不是你想用的上下文时，需要手工来配置一个上下文。设置在哈希表中的几个属性值，并且使用此哈希表来实例化一个上下文。定义语法：

Hashtable props = new Hashtable();

props.put(Context.INITIAL_CONTEXT_FACTORY,

"org.jnp.interfaces.NamingContextFactory");

props.put(Context.PROVIDER_URL, "localhost:1099");

props.put("java.naming.rmi.security.manager", "yes");

props.put(Context.URL_PKG_PREFIXES, "org.jboss.naming");

 

2) 查找连接工厂

自有了上下文后，需要查找一个连接工厂。为了查找它，使用一个可用的名字。查找连接工厂的代码如下：

对于一个topic目的地

TopicConnectionFactory topicFactory = (TopicConnectionFactory) context.lookup (“ConnectionFactory”)

Queue 目的地：

QueueConnectionFactory queueFactory = (QueueConnectionFactory ) context.lookup (“ConnectionFactory”)

3) 建立连接和会话

在我们有了连接工厂后，建立一个连接，在此连接中建立一个会话。

对于topic代码如下：

//建立一个连接

topicConnection = topicFactory.createTopicConnection();

//建立一个会话

topicSession = topicConnection.createTopicSession(false, //不需要事务

Session.AUTO_ACKNOLEDGE //自动接收消息的收条。

);

对于queue代码如下：

//建立一个连接

queueConnection = queueFactory.createQueueConnection();

//建立一个会话

queueSession = queueConnection .createQueueSession(false, //不需要事务

Session.AUTO_ACKNOLEDGE //自动接收消息的收条。

);

一个会话建立时，配置是否调用事务

在事务Session中，当事务被提交后，自动接收，如果事务回滚，所有的被消费的消息将会被重新发送。

在非事务Session中，如果没有调用事务处理，消息传递的方式有三种：

Session.AUTO_ACKNOWLEDGE :当客户机调用的receive方法成功返回，或当MessageListenser 成功处理了消息，session将会自动接收消息的收条。

Session.CLIENT_ACKNOWLEDGE ：客户机通过调用消息的acknowledge方法来接收消息。接收发生在session层。接收到一个被消费的消息时，将自动接收该session已经 消费的所有消息。例如：如果消息的消费者消费了10条消息，然后接收15个被传递的消息，则前面的10个消息的收据都会在这15个消息中被接收。

Session.DUPS_ACKNOWLEDGE ：指示session缓慢接收消息。

 

4) 查找目的地

现在我们来介绍建立publishes/sends 或subscribles/receives消息。

下面的代码列出来查找一个目的地：

对于topic 查找一个testTopic目的地

Topic topic = (Topic) context.lookup(“topic/testTopic”);

 

对于queue 查找一个testQueue目的地

Queue queue= (Queue) context.lookup(“queue/testQueue”);

注意：JbossM的前缀topic/ (queue/)通常被放在topic (queue)名字前面。

在JMS中，当客户机扮演每种角色，象对于topic来将的publisher ,subscriber 或对于queue来将的sender, receiver， 都有自己不同类继承和不同函数。

5) 建立一个消息制造者Message ProdUCer (topic publisher/ queue sender)

消息制造者是一个由session创建的对象，主要工作是发送消息到目的地。

对于一个topic，需要通过TopicSession来创建一个TopicPublisher。代码如下：

TopicPublisher topicPublisher = TopicSession.createPublisher(topic);

 

对于一个queue，需要通过QueueSession来创建一个QueueSender。代码如下：

QueuePublisher queuePublisher = queueSession.createSender(queue);

6) 消息发送

建立一个TestMessage并且publish 它， 代码：

TextMessage message = topicSession.createTestMessage();

message.setText(msg);

topicPublishe.publish(topic, message);

建立一个TestMessage并且send它， 代码：

TextMessage message = queueSession.createTestMessage();

message.setText(msg);

queueSender.send(queue, message);

 

7) 下面是一个完成的topic publisher 代码，文件名HelloPublisher.java ：

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,"org.jnp.interfaces.NamingContextFactory");

props.put(Context.PROVIDER_URL, "localhost:1099");

props.put("java.naming.rmi.security.manager", "yes");

props.put(Context.URL_PKG_PREFIXES, "org.jboss.naming");

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(

"ConnectionFactory", "topic/testTopic");

for (int i = 1; i < 11; i++) {

String msg = "Hello World no. " + i;

System.out.println("Publishing message: " + msg);

publisher.publish(msg);

}

publisher.close();

} catch(Exception ex) {

System.err.println(

"An exception occurred while testing HelloPublisher25: " + ex);

ex.printStackTrace();

}

}

}

 

我们知道，使用JMS不仅能发送（send）/发布（publish）消息，也能获得（send）/发布（publish）的消息。在时间方式有良种方法来做：

w 同步（Synchronously）：需要手工的去得到消息，为了得到一个消息客户机调用方法得到消息，直到消息到达或在规定的时间内没有到达而超时。我们在例子中没有说明这部分，大家可以实验一下。

w 异步（Asynchronously）：你需要定义一个消息监听器(MessageListener),实现该接口。当消息达到时，JMS provider通过调用该对象的 onMessage方法来传递消息。

从原则来将，topic和queue都是异步的，但是在这两种目的地中有不同的类和方法。首先，必须定义一个MessageListener接口。

8) 建立一个MessageListener

在建立了你需要的subscriber/receiver，并且登记了监听器后。就可以调用连接的start方法得到JMS provider 发送到的消息了。如果在登记监听器之前调用start方法，很可能会丢失消息。

public void onMessage(Message m) {

try {

String msg = ((TextMessage)m).getText();

System.out.println("HelloSubscriber got message: " + msg);

} catch(JMSException ex) {

System.err.println("Could not get text message: " + ex);

ex.printStackTrace();

}

}

9) 建立消息消费者

对于topic来将：

//建立一个订阅者

topicSubscriber = topicSession.createSubscriber(topic);

//设置消息监听器，

topicSubscriber.setMessageListener(this)

//连接开始

topicConnection.start();

对于queue来将：

//建立一个订阅者

queueReceiver = queueSession.createReceiver(queue);

//设置消息监听器，

queueReceiver .setMessageListener(this)

//连接开始

queueConnection.start();

10) 完整的代码，放在文件HelloSubscriber.java中，如下：

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(

"HelloSubscriber subscribed to topic: " + topicJNDI);

topicConnection.start();

}

public void onMessage(Message m) {

try {

String msg = ((TextMessage)m).getText();

System.out.println("HelloSubscriber got message: " + msg);

} catch(JMSException ex) {

System.err.println("Could not get text message: " + ex);

ex.printStackTrace();

}

}

public void close() throws JMSException {

topicSession.close();

topicConnection.close();

}

public static void main(String[] args) {

try {

HelloSubscriber subscriber = new HelloSubscriber(

"TopicConnectionFactory",

"topic/testTopic");

} catch(Exception ex) {

System.err.println(

"An exception occurred while testing HelloSubscriber: " + ex);

ex.printStackTrace();

}

}

}

 

11) 编辑、运行程序

直接使用命令（java）

w 开启命令操作符。设置classpath :

set classpath=C:jboss-3.0.6_tomcat-4.1.18clientjbossall-client.jar;C:jboss-3.0.6_tomcat-4.1.18clientjboss-j2ee.jar;C:jboss-3.0.6_tomcat-4.1.18clientjnp-client.jar;C:jboss-3.0.6_tomcat-4.1.18clientlog4j.jar;.

w 首先运行订阅消息端：java HelloSubscriber

w 再开启另外一个命令窗口设置classpath :

set classpath=C:jboss-3.0.6_tomcat-4.1.18clientjbossall-client.jar;C:jboss-3.0.6_tomcat-4.1.18clientjboss-j2ee.jar;C:jboss-3.0.6_tomcat-4.1.18clientjnp-client.jar;C:jboss-3.0.6_tomcat-4.1.18clientlog4j.jar;.

w 运行发布消息端：java HelloPublisher

5、补充

在最后我们解释JBoss-specific特性：如何用代码来管理目的地。JBoss各个版本可能不同，但是差别不大。我使用的是jboss3.0.6。

实现这个目的有两种不同的方法，依赖于是否代码是在和JBoss同样的虚拟机还是独立独立的。它们都包括调用一个通过service=DestinationManager 登记的JMX Bean。这个Mbean 有四个方法来管理目的地：createTopic()，createQueue()，destroyTopic()，destroyQueue()。

在代码中实现管理目的地在影射怎样调用MBean有不同的地方。如果程序虚拟机和Mbean服务器一样，可以直接调用。

建立一个topic 目的地的代码如下：

MBeanServer server = (MBeanServer)

MBeanServerFactory.findMBeanServer(null).iterator().next();

server.invoke(new ObjectName("JBossMQ", "service", "DestinationManager"),

method,

new Object[] { “myTopic” },

new String[] { "java.lang.String" });

 

如果程序和Mbean服务器的虚拟机不同，需要通过一个JMX adapter。一个JMX adapter是一个HTML GUI。用程序通过URL来调用Mbean。代码如下：

import java.io.InputStream;

import java.net.URL;

import java.net.HttpURLConnection;

import javax.management.MBeanServerFactory;

import javax.management.MBeanServer;

import javax.management.ObjectName;

import javax.jms.Topic;

import javax.jms.Queue;

public class DestinationHelper {

static final String HOST = "localhost";

static final int PORT = 8080;

static final String BASE_URL_ARG = "/jmx-console/HtmlAdaptor?";

public static void createDestination(Class type, String name)

throws Exception

{

String method = null;

if (type == Topic.class) { method = "createTopic"; }

else if (type == Queue.class) { method = "createQueue";}

invoke(method, name);

}

 

public static void destroyDestination(Class type, String name)

throws Exception

{

String method = null;

if (type == Topic.class) { method = "destroyTopic"; }

else if (type == Queue.class) { method = "destroyQueue";}

invoke(method, name);

}

 

protected static void invoke(String method, String destName)

throws Exception

{

try {

MBeanServer server = (MBeanServer) MBeanServerFactory.findMBeanServer(null).iterator().next();

invokeViaMBean(method, destName);

}catch(Exception ex) { invokeViaUrl(method, destName);}

}

protected static void invokeViaUrl(String method, String destName)

throws Exception

{

String action = "action=invokeOp&methodIndex=6&name=jboss.mq%3Aservice%3DDestinationManager&arg0=" + destName;

String arg = BASE_URL_ARG + action;

URL url = new URL("http", HOST, PORT, arg);

HttpURLConnection con = (HttpURLConnection)url.openConnection();

con.connect();

 

InputStream is = con.getInputStream();

java.io.ByteArrayOutputStream os = new java.io.ByteArrayOutputStream();

byte[] buff = new byte[1024];

for(;;) {

int size = is.read( buff );

if (size == -1 ) { break; }

os.write(buff, 0, size);

}

os.flush();

 

if (con.getResponseCode() != HttpURLConnection.HTTP_OK ) {

throw new Exception ("Could not invoke url: " + con.getResponseMessage() );

} else {

System.out.println("Invoked URL: " + method + " for destination " + destName + "got resonse: " + os.toString());

}

}

protected static void invokeViaMBean(String method, String destName)

throws Exception

{

MBeanServer server = (MBeanServer)MBeanServerFactory.findMBeanServer(null).iterator().next();

server.invoke(new ObjectName("JBossMQ", "service", "DestinationManager"),

method,

new Object[] { destName },

new String[] { "java.lang.String" });

}

public static void main(String[] args) {

try {

if (args.length >0){

destroyDestination(Topic.class,"myCreated");

}else {

createDestination(Topic.class,"myCreated");

}

}catch(Exception ex) {

System.out.println("Error in administering destination: " + ex);

ex.printStackTrace();

}

}

 

}

 

编辑命令：

javac -classpath C:jboss-3.0.6_tomcat-4.1.18clientjbossall-client.jar;C:jboss-3.0.6_tomcat-4.1.18libjboss-jmx.jar;. DestinationHelper.java

运行命令

java -classpath C:jboss-3.0.6_tomcat-4.1.18clientjbossall-client.jar;C:jboss-3.0.6_tomcat-4.1.18libjboss-jmx.jar;. DestinationHelper

当运行完后查看http://localhost:8080/jmx-console下面的jboss.mq.destination中有name=myCreated,service=Topic

表明你建立成功。当JBoss关闭重新启动时。该目的地不会在存在。