OpenJMS特性及使用

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

2、, 在 openjms-0.7.7-beta-1bin 里, 有 openJms 的運行腳本, 執(zhí)行 startup 啟動, 彈出一個新的窗口, 服務就運行在新窗口內, shutdown 為停止命令:服務運行后, 就可以開始使用 JMS 服務了, 至此服務搭建完畢, 簡單得不能再簡 單了。下 面 是 消 息 發(fā) 送 和 接 收 的 開 發(fā) , 開 發(fā) 中 需 要 的 jar 包 在 openjms-0.7.7-beta-1lib里可以找到:把上面的類包加入到項目中,下面是消息發(fā)送服務的代碼:/* 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默認的端口是 1099properties.put(Context.PROVIDER_URL,"rmi:/localhost:10

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

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

6、rBean(;ub.setUsername("weida"ub.setPassword("111"ObjectMessage message = queueSession.createObjectMessage(ub;queueSender.send(message;System.out.println("信息寫入 JMS 服務器隊列 "/以下做清除工作,代碼略/ . . 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 信息連接隊列工廠QueueConnectionFactory queueConnectionFactory =(QueueConnectionFactory context.lookup("JmsQueueConnectionFactory"/獲得 JMS 信息連接隊列QueueConnection queueConnection =queueConnectionFactory.createQueueConnection(;/啟動接收隊

9、列線程queueConnection.start(;/產生隊列 Session ,設置事務為 false ,自動應答消息接收 QueueSession queueSession =queueConnection.createQueueSession(false,Session.AUTO_ACKNOWLEDGE;/獲得默認內建在 JMS 里的隊列之一:queue1Queue queue = (Queue context.lookup("queue1"/產生 JMS 隊列接收器QueueReceiver queueReceiver =queueSession.createRec

10、eiver(queue;/通過同步的方法接收消息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 信息連接隊列工廠QueueConnectionFactory queueConnectionFactory =(QueueConnectionFactory context.lookup("

13、;JmsQueueConnectionFactory"/獲得 JMS 信息連接隊列QueueConnection queueConnection =queueConnectionFactory.createQueueConnection(; /產生隊列 Session ,設置事務為 false ,自動應答消息接收 QueueSession queueSession =queueConnection.createQueueSession(false,Session.AUTO_ACKNOWLEDGE;/獲得默認內建在 JMS 里的隊列之一:queue1Queue queue = (Que

14、ue context.lookup("queue1"/產生 JMS 隊列接收器QueueReceiver queueReceiver =queueSession.createReceiver(queue;queueReceiver.setMessageListener(this;/啟動接收隊列線程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 連接隊列工廠 TopicConnectionFa

19、ctory factory = (TopicConnectionFactory context .lookup("JmsTopicConnectionFactory" / 創(chuàng)建一個 Topic 連接，并啟動 TopicConnection topicConnection factory.createTopicConnection(; topicConnection.start(; / 創(chuàng)建一個 Topic 會話，并設置自動應答 m_topicSession = topicConnection.createTopicSession(false, Session.AUTO_AC

20、KNOWLEDGE; / lookup 得到 topic1 Topic topic = (Topic context.lookup(strtopic; / 用 Topic 會話生成 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("請先初始化 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("定購消息接收啟動：" /取得 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 的會話，用于接收信息 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("訂閱接收結束." 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"