OpenJMS特性及使用

1、OpenJMS 是一個(gè)開(kāi)源的 Java Message Service API 1.0.2 規(guī)范的實(shí)現(xiàn) , 它包含有以下特 性:*. 它既支持點(diǎn)到點(diǎn)(point-to-point (PTP 模型和發(fā)布 /訂閱(Pub/Sub模型。 *. 支持同步與異步消息發(fā)送*. JDBC持久性管理使用數(shù)據(jù)庫(kù)表來(lái)存儲(chǔ)消息*. 可視化管理界面。*. Applet支持。*. 能夠與 Jakarta Tomcat這樣的 Servlet 容器結(jié)合。*. 支持 RMI, TCP, HTTP 與 SSL 協(xié)議。*. 客戶端驗(yàn)證*. 提供可靠消息傳輸、事務(wù)和消息過(guò)濾介紹服務(wù)器搭建,在 openJms , 解壓后可以直接使用

2、, 在 openjms-0.7.7-beta-1bin 里, 有 openJms 的運(yùn)行腳本, 執(zhí)行 startup 啟動(dòng), 彈出一個(gè)新的窗口, 服務(wù)就運(yùn)行在新窗口內(nèi), shutdown 為停止命令:服務(wù)運(yùn)行后, 就可以開(kāi)始使用 JMS 服務(wù)了, 至此服務(wù)搭建完畢, 簡(jiǎn)單得不能再簡(jiǎn) 單了。下 面 是 消 息 發(fā) 送 和 接 收 的 開(kāi) 發(fā) , 開(kāi) 發(fā) 中 需 要 的 jar 包 在 openjms-0.7.7-beta-1lib里可以找到:把上面的類包加入到項(xiàng)目中,下面是消息發(fā)送服務(wù)的代碼:/* author qianli 2011-12-26* For openJms 演示 , Messag

3、e 發(fā)送*/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)的端口是 1099properties.put(Context.PROVIDER_URL,"rmi:/localhost:10

4、99/"Context context = new InitialContext(properties;/獲得 JMS 信息連接隊(duì)列工廠QueueConnectionFactory queueConnectionFactory =(QueueConnectionFactory context.lookup("JmsQueueConnectionFactory"/獲得 JMS 信息連接隊(duì)列QueueConnection queueConnection =queueConnectionFactory.createQueueConnection(;/產(chǎn)生隊(duì)列 Sessi

5、on ,設(shè)置事務(wù)為 false ,自動(dòng)應(yīng)答消息接收 QueueSession queueSession =queueConnection.createQueueSession(false,Session.AUTO_ACKNOWLEDGE;/獲得默認(rèn)內(nèi)建在 JMS 里的隊(duì)列之一:queue1Queue queue = (Queue context.lookup("queue1"/產(chǎn)生 JMS 隊(duì)列發(fā)送器QueueSender queueSender =queueSession.createSender(queue;/發(fā)送數(shù)據(jù)到 JMSUserBean ub = new Use

6、rBean(;ub.setUsername("weida"ub.setPassword("111"ObjectMessage message = queueSession.createObjectMessage(ub;queueSender.send(message;System.out.println("信息寫(xiě)入 JMS 服務(wù)器隊(duì)列 "/以下做清除工作,代碼略/ . . catch (Exception e e.printStackTrace(;/* author qianli 2011-12-26* For openJms 演示

7、, 同步接收 Message*/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:/localho

8、st:1099/"Context context = new InitialContext(properties;/獲得 JMS 信息連接隊(duì)列工廠QueueConnectionFactory queueConnectionFactory =(QueueConnectionFactory context.lookup("JmsQueueConnectionFactory"/獲得 JMS 信息連接隊(duì)列QueueConnection queueConnection =queueConnectionFactory.createQueueConnection(;/啟動(dòng)接收隊(duì)

9、列線程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ì)列之一:queue1Queue queue = (Queue context.lookup("queue1"/產(chǎn)生 JMS 隊(duì)列接收器QueueReceiver queueReceiver =queueSession.createRec

10、eiver(queue;/通過(guò)同步的方法接收消息Message message = queueReceiver.receive(;UserBean ub = null;if (message instanceof ObjectMessageub = (UserBean(ObjectMessagemessage.getObject(; System.out.println("username="+ub.getUsername(; System.out.println("password="+ub.getPassword(; /以下做清除工作,代碼略/ .

11、. catch (Exception e e.printStackTrace(;/* author qianli 2011-12-26* For openJms 演示 , 非同步接收 Message*/public class QueueReceiveAsynchronous implements MessageListener private UserBean ub = null;QueueReceiveAsynchronous(try /取得 JNDI 上下文和連接Hashtable properties = new Hashtable(;properties.put(Context.IN

12、ITIAL_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("

13、;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ì)列之一:queue1Queue queue = (Que

14、ue context.lookup("queue1"/產(chǎn)生 JMS 隊(duì)列接收器QueueReceiver queueReceiver =queueSession.createReceiver(queue;queueReceiver.setMessageListener(this;/啟動(dòng)接收隊(duì)列線程queueConnection.start(; catch (Exception e e.printStackTrace(;public static void main(String args System.out.println("非同步消息的接收:"try

15、 QueueReceiveAsynchronous listener =new QueueReceiveAsynchronous(;Thread.sleep(1000; catch (InterruptedException e e.printStackTrace(;public void onMessage(Message message if (message instanceof ObjectMessagetry ub = (UserBean(ObjectMessagemessage.getObject(; catch (JMSException e e.printStackTrace(

16、;if(ub != nullSystem.out.println("username="+ub.getUsername(; System.out.println("password="+ub.getPassword(; ub = null;/* author qianli 2011-12-26* openJms 訂閱 /發(fā)布模式 發(fā)布*/public class TopicSender private TopicPublisher m_topicPublisher;private TopicSession m_topicSession;private i

17、nt binit = 0;public TopicSender(String strtopic try Hashtable properties = new Hashtable(;properties.put(Context.INITIAL_CONTEXT_FACTORY,"org.exolab.jms.jndi.InitialContextFactory"properties.put(Context.PROVIDER_URL, "rmi:/localhost:1099/" /*properties.put(".https.keyStore&q

18、uot;,"E:worksnote_namesrcclient.keystore"properties.put(".https.keyStorePassword","openjms" properties.put(".https.trustStore", "C:jsdk2jrelibsecuritycacerts" */ Context context = new InitialContext(properties; / 獲得 JMS Topic 連接隊(duì)列工廠 TopicConnectionFa

19、ctory 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)答 m_topicSession = topicConnection.createTopicSession(false, Session.AUTO_AC

20、KNOWLEDGE; / lookup 得到 topic1 Topic topic = (Topic context.lookup(strtopic; / 用 Topic 會(huì)話生成 Topic 發(fā)布器 m_topicPublisher = m_topicSession.createPublisher(topic; binit = 1; catch (NamingException e e.printStackTrace(; catch (JMSException e e.printStackTrace(; public void Send(String strmsg if (binit = 0

21、 System.out.println("請(qǐng)先初始化 jms" else try TextMessage m_topicSession.createTextMessage(strmsg; m_topicPublisher.publish(message; catch (JMSException e e.printStackTrace(; = message = public static void main(String args TopicSender send = new TopicSender("topic1" send.Send("he

22、llo" /* * author qianli 2011-12-26 * openJms 非持久訂閱同步接收演示 */ 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.ex

23、olab.jms.jndi.InitialContextFactory" properties.put(Context.PROVIDER_URL, "rmi:/localhost:1099/" Context context = new InitialContext(properties; /獲得 Topic 工廠和 Connection TopicConnectionFactory factory = (TopicConnectionFactory context.lookup( "JmsTopicConnectionFactory" Top

24、icConnection 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 TopicS

25、ubscriber topicSubscriber = topicSession.createSubscriber(topic; /收滿 10 條訂閱消息則退出 for (int i=0; i<10; i+ /同步消息接收，使用 receive 方法，堵塞等待，直到接收消 息 = TextMessage message = (TextMessage topicSubscriber.receive(; System.out.println(" 接 收 訂 閱 消 息 "+i+": " + message.getText(; /資源清除，代碼略 . .

26、 System.out.println("訂閱接收結(jié)束." catch (NamingException e e.printStackTrace(; catch (JMSException e e.printStackTrace(; /* * author qianli 2011-12-26 * openJms 非持久訂閱異步接收演示 */ public class TopicSubscribeAsynchronous implements MessageListener private TopicConnection topicConnection; private To

27、picSession 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"