細(xì)心!用心!耐心!

          吾非文人,乃市井一俗人也,讀百卷書,跨江河千里,故申城一游; 一兩滴辛酸,三四年學(xué)業(yè),五六點粗墨,七八筆買賣,九十道人情。

          BlogJava 聯(lián)系 聚合 管理
            1 Posts :: 196 Stories :: 10 Comments :: 0 Trackbacks

          JMS服務(wù)器openJms入門

          首先可以到網(wǎng)站上下載最新的openjms版本,然后啟動bin下的startup.bat就啟動服務(wù)了,然后可以啟動admin.bat進入管理界面(非必要步驟),然后分別運行samples下的basic例子就可以看到效果了,當(dāng)然以前有篇文章介紹的不錯,可以參考一下,不過時間上比較老了,做為參考還是可以的。

          JMS 有五種消息類型。三種結(jié)構(gòu)化或半結(jié)構(gòu)化的消息類型(MapMessage、ObjectMessage 和 StreamMessage)以及兩種非結(jié)構(gòu)化的或自由格式的消息類型(TextMessage 和 BytesMessage)。而這里雖然我們只對非結(jié)構(gòu)化消息進行說明,但非結(jié)構(gòu)化的消息格式卻能夠更好地進行交互操作,因為它們在消息上很少利用結(jié)構(gòu),在此基礎(chǔ)上與XML再進行結(jié)合,將能方便的進行更好的擴展,XML相關(guān)簡化操作參考《Jaxb來實現(xiàn)Java-XML的轉(zhuǎn)換》。

          下面具體來介紹服務(wù)器搭建,在http://openjms.sourceforge.net/downloads.html下載openJms,解壓后可以直接使用,在 \openjms-0.7.6.1\bin 里,有openJms的運行腳本,執(zhí)行 startup 啟動,彈出一個新的窗口,服務(wù)就運行在新窗口內(nèi),shutdown 為停止命令:

          服務(wù)運行后,就可以開始使用JMS服務(wù)了,至此服務(wù)搭建完畢,簡單得不能再簡單了。

          下面是消息發(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

          把上面的類包加入到項目中,下面是消息發(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信息連接隊列工廠
                      QueueConnectionFactory queueConnectionFactory =
                          (QueueConnectionFactory) context.lookup(
                              "JmsQueueConnectionFactory");
                      //獲得JMS信息連接隊列
                      QueueConnection queueConnection =
                          queueConnectionFactory.createQueueConnection();
                      //產(chǎn)生隊列Session,設(shè)置事務(wù)為false,自動應(yīng)答消息接收
                      QueueSession queueSession =
                          queueConnection.createQueueSession(
                              false,
                              Session.AUTO_ACKNOWLEDGE);

                      //獲得默認(rèn)內(nèi)建在JMS里的隊列之一:queue1
                      Queue queue = (Queue) context.lookup("queue1");
                      //產(chǎn)生JMS隊列發(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ù)器隊列");

                      //以下做清除工作,代碼略
                      // ... ...
                                
                  } catch (Exception e) {
                      e.printStackTrace();
                  }
              }
          }

          執(zhí)行程序發(fā)送消息,然后打開JMS控制臺,用 admin 命令啟動管理平臺,點擊菜單Actions-Connections-online,出現(xiàn)界面如下:

          可以看到JSM默認(rèn)的隊列queue1里已經(jīng)有1條消息了,而其他的隊列還是空著的。

          下面我們來看看消息接收服務(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信息連接隊列工廠
                      QueueConnectionFactory queueConnectionFactory =
                          (QueueConnectionFactory) context.lookup(
                              "JmsQueueConnectionFactory");

                      //獲得JMS信息連接隊列
                      QueueConnection queueConnection =
                          queueConnectionFactory.createQueueConnection();

                      //啟動接收隊列線程
                      queueConnection.start();
                      //產(chǎn)生隊列Session,設(shè)置事務(wù)為false,自動應(yīng)答消息接收
                      QueueSession queueSession =
                          queueConnection.createQueueSession(
                              false,
                              Session.AUTO_ACKNOWLEDGE);
                      //獲得默認(rèn)內(nèi)建在JMS里的隊列之一:queue1
                      Queue queue = (Queue) context.lookup("queue1");
                      //產(chǎn)生JMS隊列接收器
                      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();
                  }
              }
          }

          編譯后運行接收信息服務(wù),可以看到接收到并打印之前發(fā)送的消息,再看看控制臺,發(fā)現(xiàn)queue1的消息隊列變?yōu)?,消息已被讀取,消息發(fā)送和接收到此結(jié)束。

           

          上篇openJms介紹 (一)  提到了openJms的構(gòu)建及消息的發(fā)送和接收,這篇主要了解消息的發(fā)布和訂閱。JMS 的發(fā)布/訂閱模型定義了如何向一個內(nèi)容節(jié)點發(fā)布和訂閱消息,內(nèi)容節(jié)點也叫主題(topic),主題是為發(fā)布者(publisher)和訂閱者 (subscribe) 提供傳輸?shù)闹薪椤0l(fā)布/訂閱模型使發(fā)布者和訂閱者之間不需要直接通訊(如RMI)就可保證消息的傳送,有效解決系統(tǒng)間耦合問題(當(dāng)然有這個需要才行),還有就是提供了一對一、一對多的通訊方式,比較靈活。

          先介紹JMS里2個概念,持久訂閱模式和非持久訂閱模式,其實也是發(fā)布/訂閱模型在可靠性上提供的2種方式:

          非持久訂閱模式:只有當(dāng)客戶端處于激活狀態(tài),也就是和JMS 服務(wù)器保持連接的狀態(tài)下,才能接收到發(fā)送到某個Topic的消息,而當(dāng)客戶端處于離線狀態(tài)時,則這個時間段發(fā)到Topic的消息將會永遠(yuǎn)接收不到。

          持久訂閱模式:客戶端向JMS 注冊一個識別自己身份的ID,當(dāng)這個客戶端處于離線時,JMS 服務(wù)器會為這個ID 保存所有發(fā)送到主題的消息,當(dāng)客戶再次連接到JMS 服務(wù)器時,會根據(jù)自己的ID 得到所有當(dāng)自己處于離線時發(fā)送到主題的消息,即消息永遠(yuǎn)能接收到。

          下面我們就接著來看openJms在發(fā)布/訂閱模式上的表現(xiàn),由于篇幅關(guān)系,在這里只講述非持久訂閱模式,持久訂閱模式可以根據(jù)JMS的標(biāo)準(zhǔn)來試。

          消息發(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連接隊列工廠
                      TopicConnectionFactory factory =
                          (TopicConnectionFactory) context.lookup(
                              "JmsTopicConnectionFactory");

                      //創(chuàng)建一個Topic連接,并啟動
                      TopicConnection topicConnection = factory.createTopicConnection();
                      topicConnection.start();

                      //創(chuàng)建一個Topic會話,并設(shè)置自動應(yīng)答
                      TopicSession topicSession =
                          topicConnection.createTopicSession(false,
                                  Session.AUTO_ACKNOWLEDGE);

                      //lookup 得到 topic1
                      Topic topic = (Topic) context.lookup("topic1");
                      //用Topic會話生成Topic發(fā)布器
                      TopicPublisher topicPublisher = topicSession.createPublisher(topic);

                      //發(fā)布消息到Topic
                      System.out.println("消息發(fā)布到Topic");
                      TextMessage message = topicSession.createTextMessage
                          ("你好,歡迎定購Topic類消息");
                      topicPublisher.publish(message);

                      //資源清除,代碼略  ... ...   
                  } catch (NamingException e) {
                      e.printStackTrace();
                  } catch (JMSException e) {
                      e.printStackTrace();
                  }
              }
          }

          而訂閱消息的接收有同步的和異步2種,他們分別使用receive()和onMessage(Message message)方法來接收消息,具體代碼:

          同步接收:

          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("定購消息接收啟動:");
                      //取得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的會話,用于接收信息
                      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的會話,用于接收信息
                      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);

                      //啟動信息接收
                      topicConnection.start();
                  } catch (NamingException e) {
                      e.printStackTrace();
                  } catch (JMSException e) {
                      e.printStackTrace();
                  }
              }

              public static void main(String[] args) {
                  System.out.println("非同步定購消息的接收:");
                  try {
                      TopicSubscribeAsynchronous listener =
                          new TopicSubscribeAsynchronous();
                      Thread.sleep(1000);
                  } catch (InterruptedException e) {
                      e.printStackTrace();
                  }
              }

              //收到訂閱信息后自動調(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();
                  }
              }
          }

          編譯好后,啟動openJms服務(wù),打開admin管理臺,為了運行方便,這里先列出三個類的運行命令:
          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

          先運行2個接收命令,再運行發(fā)布命令,可以看到控制臺的Topic有消息接收,并且接收1和2都有消息接收的提示,到此完成演示,由于是非持久訂閱,所以可以看到控制臺上的Topic消息條數(shù)不會減少。

          posted on 2007-05-06 11:45 張金鵬 閱讀(156) 評論(0)  編輯  收藏

          只有注冊用戶登錄后才能發(fā)表評論。


          網(wǎng)站導(dǎo)航:
           
          主站蜘蛛池模板: 苗栗市| 上饶县| 崇仁县| 巴东县| 珲春市| 林州市| 巴中市| 广水市| 彰化县| 赫章县| 广宁县| 神农架林区| 小金县| 石城县| 西藏| 汤阴县| 乌审旗| 南召县| 精河县| 泗洪县| 阿拉尔市| 临泉县| 大埔县| 西充县| 朔州市| 满洲里市| 酒泉市| 根河市| 崇礼县| 铁力市| 贺州市| 孝感市| 锦屏县| 旬邑县| 大埔县| 淄博市| 青浦区| 朝阳县| 梁河县| 青阳县| 鲜城|