首
先可以到網(wǎng)站上下載最新的openjms版本，然后啟動(dòng)bin下的startup.bat就啟動(dòng)服務(wù)了，然后可以啟動(dòng)admin.bat進(jìn)入管理界面（非
必要步驟），然后分別運(yùn)行samples下的basic例子就可以看到效果了，當(dāng)然以前有篇文章介紹的不錯(cuò)，可以參考一下，不過時(shí)間上比較老了，做為參考
還是可以的。
本文介紹開源的JMS服務(wù)器openJms，及怎樣使用openJms來(lái)構(gòu)建系統(tǒng)之間健全、高度可用的通訊，從而簡(jiǎn)化企業(yè)級(jí)應(yīng)用的開發(fā)。
openJms符合SUN的JMS API
1.0.2規(guī)范，支持消息隊(duì)列，還支持消息傳遞的發(fā)布/訂閱模式，本文先就系統(tǒng)服務(wù)的搭建及JMS的非結(jié)構(gòu)化消息發(fā)送和接收進(jìn)行說(shuō)明。
JMS 有五種消息類型。三種結(jié)構(gòu)化或半結(jié)構(gòu)化的消息類型（MapMessage、ObjectMessage 和
StreamMessage）以及兩種非結(jié)構(gòu)化的或自由格式的消息類型（TextMessage 和
BytesMessage）。而這里雖然我們只對(duì)非結(jié)構(gòu)化消息進(jìn)行說(shuō)明，但非結(jié)構(gòu)化的消息格式卻能夠更好地進(jìn)行交互操作，因?yàn)樗鼈冊(cè)谙⑸虾苌倮媒Y(jié)構(gòu)，
在此基礎(chǔ)上與XML再進(jìn)行結(jié)合，將能方便的進(jìn)行更好的擴(kuò)展，XML相關(guān)簡(jiǎn)化操作參考《Jaxb來(lái)實(shí)現(xiàn)Java-XML的轉(zhuǎn)換》。
下面具體來(lái)介紹服務(wù)器搭建，在http://openjms.sourceforge.net/downloads.html下載openJms，解壓后
可以直接使用，在 "openjms-0.7.6.1"bin 里，有openJms的運(yùn)行腳本，執(zhí)行 startup
啟動(dòng)，彈出一個(gè)新的窗口，服務(wù)就運(yùn)行在新窗口內(nèi)，shutdown 為停止命令：
服務(wù)運(yùn)行后，就可以開始使用JMS服務(wù)了，至此服務(wù)搭建完畢，簡(jiǎn)單得不能再簡(jiǎn)單了。
下面是消息發(fā)送和接收的開發(fā)，開發(fā)中需要的jar包在"openjms-0.7.6.1"lib里可以找到：
openjms-0.7.6.1.jar
jms-1.0.2a.jar
exolabcore-0.3.7.jar
commons-logging-1.0.3.jar
把上面的類包加入到項(xiàng)目中，下面是消息發(fā)送服務(wù)的代碼：
package javayou.demo.openjms;
import java.util.*;
import javax.jms.*;
import javax.naming.*;
public class QueueSend {
public static void main(String[] args) {
try {
//取得JNDI上下文和連接
Hashtable properties = new Hashtable();
properties.put(
Context.INITIAL_CONTEXT_FACTORY,
"org.exolab.jms.jndi.InitialContextFactory");
//openJms默認(rèn)的端口是1099
properties.put(Context.PROVIDER_URL,
"rmi://localhost:1099/");
Context context = new InitialContext(properties);
//獲得JMS信息連接隊(duì)列工廠
QueueConnectionFactory queueConnectionFactory =
(QueueConnectionFactory) context.lookup(
"JmsQueueConnectionFactory");
//獲得JMS信息連接隊(duì)列
QueueConnection queueConnection =
queueConnectionFactory.createQueueConnection();
//產(chǎn)生隊(duì)列Session，設(shè)置事務(wù)為false，自動(dòng)應(yīng)答消息接收
QueueSession queueSession =
queueConnection.createQueueSession(
false,
Session.AUTO_ACKNOWLEDGE);
//獲得默認(rèn)內(nèi)建在JMS里的隊(duì)列之一：queue1
Queue queue = (Queue) context.lookup("queue1");
//產(chǎn)生JMS隊(duì)列發(fā)送器
QueueSender queueSender =
queueSession.createSender(queue);
//發(fā)送數(shù)據(jù)到JMS
TextMessage message = queueSession.createTextMessage();
message.setText("Hello, I'm openJms.");
queueSender.send(message);
System.out.println(
""信息寫入JMS服務(wù)器隊(duì)列");
//以下做清除工作，代碼略
// ... ...
} catch (Exception e) {
e.printStackTrace();
}
}
}
執(zhí)行程序發(fā)送消息，然后打開JMS控制臺(tái)，用 admin 命令啟動(dòng)管理平臺(tái)，點(diǎn)擊菜單Actions－Connections－online，出現(xiàn)界面如下：
可以看到JSM默認(rèn)的隊(duì)列queue1里已經(jīng)有1條消息了，而其他的隊(duì)列還是空著的。
下面我們來(lái)看看消息接收服務(wù)的代碼：
package javayou.demo.openjms;
import java.util.*;
import javax.jms.*;
import javax.naming.*;
public class QueueReceiveSynchronous {
public static void main(String[] args) {
try {
//取得JNDI上下文和連接
Hashtable properties = new Hashtable();
properties.put(
Context.INITIAL_CONTEXT_FACTORY,
"org.exolab.jms.jndi.InitialContextFactory");
properties.put(Context.PROVIDER_URL,
"rmi://localhost:1099/");
Context context = new InitialContext(properties);
//獲得JMS信息連接隊(duì)列工廠
QueueConnectionFactory queueConnectionFactory =
(QueueConnectionFactory) context.lookup(
"JmsQueueConnectionFactory");
//獲得JMS信息連接隊(duì)列
QueueConnection queueConnection =
queueConnectionFactory.createQueueConnection();
//啟動(dòng)接收隊(duì)列線程
queueConnection.start();
//產(chǎn)生隊(duì)列Session，設(shè)置事務(wù)為false，自動(dòng)應(yīng)答消息接收
QueueSession queueSession =
queueConnection.createQueueSession(
false,
Session.AUTO_ACKNOWLEDGE);
//獲得默認(rèn)內(nèi)建在JMS里的隊(duì)列之一：queue1
Queue queue = (Queue) context.lookup("queue1");
//產(chǎn)生JMS隊(duì)列接收器
QueueReceiver queueReceiver =
queueSession.createReceiver(queue);
//通過同步的方法接收消息
Message message = queueReceiver.receive();
String messageText = null;
if (message instanceof TextMessage)
messageText = ((TextMessage) message).
getText();
System.out.println(messageText);
//以下做清除工作，代碼略
// ... ...
} catch (Exception e) {
e.printStackTrace();
}
}
}
編譯后運(yùn)行接收信息服務(wù)，可以看到接收到并打印之前發(fā)送的消息，再看看控制臺(tái)，發(fā)現(xiàn)queue1的消息隊(duì)列變?yōu)?，消息已被讀取，消息發(fā)送和接收到此結(jié)束。
上篇openJms介紹 (一) 提到了openJms的構(gòu)建及消息的發(fā)送和接收，這篇主要了解消息的發(fā)布和訂閱。JMS
的發(fā)布/訂閱模型定義了如何向一個(gè)內(nèi)容節(jié)點(diǎn)發(fā)布和訂閱消息，內(nèi)容節(jié)點(diǎn)也叫主題(topic)，主題是為發(fā)布者(publisher)和訂閱者
(subscribe)
提供傳輸?shù)闹薪椤０l(fā)布/訂閱模型使發(fā)布者和訂閱者之間不需要直接通訊（如RMI）就可保證消息的傳送，有效解決系統(tǒng)間耦合問題（當(dāng)然有這個(gè)需要才行），還
有就是提供了一對(duì)一、一對(duì)多的通訊方式，比較靈活。
先介紹JMS里2個(gè)概念，持久訂閱模式和非持久訂閱模式，其實(shí)也是發(fā)布/訂閱模型在可靠性上提供的2種方式：
非持久訂閱模式：只有當(dāng)客戶端處于激活狀態(tài)，也就是和JMS 服務(wù)器保持連接的狀態(tài)下，才能接收到發(fā)送到某個(gè)Topic的消息，而當(dāng)客戶端處于離線狀態(tài)時(shí)，則這個(gè)時(shí)間段發(fā)到Topic的消息將會(huì)永遠(yuǎn)接收不到。
持久訂閱模式：客戶端向JMS 注冊(cè)一個(gè)識(shí)別自己身份的ID，當(dāng)這個(gè)客戶端處于離線時(shí)，JMS 服務(wù)器會(huì)為這個(gè)ID 保存所有發(fā)送到主題的消息，當(dāng)客戶再次連接到JMS 服務(wù)器時(shí)，會(huì)根據(jù)自己的ID 得到所有當(dāng)自己處于離線時(shí)發(fā)送到主題的消息，即消息永遠(yuǎn)能接收到。
下面我們就接著來(lái)看openJms在發(fā)布/訂閱模式上的表現(xiàn)，由于篇幅關(guān)系，在這里只講述非持久訂閱模式，持久訂閱模式可以根據(jù)JMS的標(biāo)準(zhǔn)來(lái)試。
消息發(fā)布的代碼如下：
package javayou.demo.openjms;
import java.util.*;
import javax.jms.*;
import javax.naming.*;
public class TopicPublish {
public static void main(String[] args) {
try {
//取得JNDI上下文和連接
Hashtable properties = new Hashtable();
properties.put(
Context.INITIAL_CONTEXT_FACTORY,
"org.exolab.jms.jndi.InitialContextFactory");
//openJms默認(rèn)的端口是1099
properties.put(Context.PROVIDER_URL, "rmi://localhost:1099/");
Context context = new InitialContext(properties);
//獲得JMS Topic連接隊(duì)列工廠
TopicConnectionFactory factory =
(TopicConnectionFactory) context.lookup(
"JmsTopicConnectionFactory");
//創(chuàng)建一個(gè)Topic連接，并啟動(dòng)
TopicConnection topicConnection = factory.createTopicConnection();
topicConnection.start();
//創(chuàng)建一個(gè)Topic會(huì)話，并設(shè)置自動(dòng)應(yīng)答
TopicSession topicSession =
topicConnection.createTopicSession(false,
Session.AUTO_ACKNOWLEDGE);
//lookup 得到 topic1
Topic topic = (Topic) context.lookup("topic1");
//用Topic會(huì)話生成Topic發(fā)布器
TopicPublisher topicPublisher = topicSession.createPublisher(topic);
//發(fā)布消息到Topic
System.out.println("消息發(fā)布到Topic");
TextMessage message = topicSession.createTextMessage
("你好，歡迎定購(gòu)Topic類消息");
topicPublisher.publish(message);
//資源清除，代碼略 ... ...
} catch (NamingException e) {
e.printStackTrace();
} catch (JMSException e) {
e.printStackTrace();
}
}
}
而訂閱消息的接收有同步的和異步2種，他們分別使用receive()和onMessage(Message message)方法來(lái)接收消息，具體代碼：
同步接收：
package javayou.demo.openjms;
import java.util.*;
import javax.jms.*;
import javax.naming.*;
public class TopicSubscribeSynchronous {
public static void main(String[] args) {
try {
System.out.println("定購(gòu)消息接收啟動(dòng)：");
//取得JNDI上下文和連接
Hashtable properties = new Hashtable();
properties.put(Context.INITIAL_CONTEXT_FACTORY,
"org.exolab.jms.jndi.InitialContextFactory");
properties.put(Context.PROVIDER_URL, "rmi://localhost:1099/");
Context context = new InitialContext(properties);
//獲得Topic工廠和Connection
TopicConnectionFactory factory =
(TopicConnectionFactory) context.lookup(
"JmsTopicConnectionFactory");
TopicConnection topicConnection = factory.createTopicConnection();
topicConnection.start();
//創(chuàng)建Topic的會(huì)話，用于接收信息
TopicSession topicSession =
topicConnection.createTopicSession(
false,
Session.AUTO_ACKNOWLEDGE);
//lookup topic1
Topic topic = (Topic) context.lookup("topic1");
//創(chuàng)建Topic subscriber
TopicSubscriber topicSubscriber =
topicSession.createSubscriber(topic);
//收滿10條訂閱消息則退出
for (int i=0; i<10; i++) {
//同步消息接收，使用receive方法，堵塞等待，直到接收消息
TextMessage message = (TextMessage) topicSubscriber.receive();
System.out.println("接收訂閱消息["+i+"]: " + message.getText());
}
//資源清除，代碼略 ... ...
System.out.println("訂閱接收結(jié)束.");
} catch (NamingException e) {
e.printStackTrace();
} catch (JMSException e) {
e.printStackTrace();
}
}
}
非同步接收：
package javayou.demo.openjms;
import java.util.*;
import javax.jms.*;
import javax.naming.*;
public class TopicSubscribeAsynchronous implements MessageListener {
private TopicConnection topicConnection;
private TopicSession topicSession;
private Topic topic;
private TopicSubscriber topicSubscriber;
TopicSubscribeAsynchronous() {
try {
//取得JNDI上下文和連接
Hashtable properties = new Hashtable();
properties.put(
Context.INITIAL_CONTEXT_FACTORY,
"org.exolab.jms.jndi.InitialContextFactory");
properties.put(Context.PROVIDER_URL, "rmi://localhost:1099/");
Context context = new InitialContext(properties);
//取得Topic的連接工廠和連接
TopicConnectionFactory topicConnectionFactory =
(TopicConnectionFactory) context.lookup(
"JmsTopicConnectionFactory");
topicConnection = topicConnectionFactory.createTopicConnection();
//創(chuàng)建Topic的會(huì)話，用于接收信息
topicSession =
topicConnection.createTopicSession(false,
Session.AUTO_ACKNOWLEDGE);
topic = (Topic) context.lookup("topic1");
//創(chuàng)建Topic subscriber
topicSubscriber = topicSession.createSubscriber(topic);
//設(shè)置訂閱監(jiān)聽
topicSubscriber.setMessageListener(this);
//啟動(dòng)信息接收
topicConnection.start();
} catch (NamingException e) {
e.printStackTrace();
} catch (JMSException e) {
e.printStackTrace();
}
}
public static void main(String[] args) {
System.out.println("非同步定購(gòu)消息的接收：");
try {
TopicSubscribeAsynchronous listener =
new TopicSubscribeAsynchronous();
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//收到訂閱信息后自動(dòng)調(diào)用此方法
public void onMessage(Message message) {
try {
String messageText = null;
if (message instanceof TextMessage)
messageText = ((TextMessage) message).getText();
System.out.println(messageText);
} catch (JMSException e) {
e.printStackTrace();
}
}
}
編譯好后，啟動(dòng)openJms服務(wù)，打開admin管理臺(tái)，為了運(yùn)行方便，這里先列出三個(gè)類的運(yùn)行命令：
java -cp ."; -Djava.ext.dirs=."lib; javayou.demo.openjms.TopicPublish
java -cp ."; -Djava.ext.dirs=."lib; javayou.demo.openjms.TopicSubscribeSynchronous
java -cp ."; -Djava.ext.dirs=."lib; javayou.demo.openjms.TopicSubscribeAsynchronous
先運(yùn)行2個(gè)接收命令，再運(yùn)行發(fā)布命令，可以看到控制臺(tái)的Topic有消息接收，并且接收1和2都有消息接收的提示，到此完成演示，由于是非持久訂閱，所以可以看到控制臺(tái)上的Topic消息條數(shù)不會(huì)減少。