`
seaizon
  • 浏览: 138053 次
  • 性别: Icon_minigender_1
  • 来自: 北京
社区版块
存档分类
最新评论

使用activeMQ实现jms(转)

    博客分类:
  • java
阅读更多

 

 

 

一:jms介绍  
         jms说白了就是java message service,是J2EE规范的一部分,跟jdbc差不多,sun只提供了接口,由各个厂商(provider)来进行具体的实现,然后使用者使用他们的jar包进行开发使用即可。
        另外在jms的API中,jms传递消息有两种方式,一种是点对点的Queue,还有一个是发布订阅的Topic方式。区别在于:
        对于Queue模式,一个发布者发布消息,下面的接收者按队列顺序接收,比如发布了10个消息,两个接收者A,B那就是A,B总共会收到10条消息,不重复。
        对于Topic模式,一个发布者发布消息,有两个接收者A,B来订阅,那么发布了10条消息,A,B各收到10条消息。
       关于api的简单基础可以看下:http://www.iteye.com/topic/64707,简单的参考!
二:ActiveMQ介绍  
         activeMQ是apache下的一个开源jms产品,具体参见apache官方网站;  
         Apache ActiveMQ is fast, supports many    Cross Language Clients and Protocols   , comes with easy to use  Enterprise Integration Patterns       and many    advanced features       while fully supporting    JMS 1.1       and J2EE 1.4. Apache ActiveMQ is released under the    Apache       2.0 License  
三:开始实现代码  
       1: 使用activeMQ来完成jms的发送,必须要下载activeMQ,然后再本机安装,并且启动activeMQ的服务才行。在官网下载完成之后,运行bin目录下面的activemq.bat,将activeMQ成功启动。
       启动成功之后可以运行:http://localhost:8161/admin/index.jsp  查看一下。
       2:发送端,sender

 

Java代码 
  1. import  javax.jms.Connection;  
  2. import  javax.jms.ConnectionFactory;  
  3. import  javax.jms.DeliveryMode;  
  4. import  javax.jms.Destination;  
  5. import  javax.jms.MessageProducer;  
  6. import  javax.jms.Session;  
  7. import  javax.jms.TextMessage;  
  8.   
  9. import  org.apache.activemq.ActiveMQConnection;  
  10. import  org.apache.activemq.ActiveMQConnectionFactory;  
  11.   
  12. public   class  Sender {  
  13.     private   static   final   int  SEND_NUMBER =  5 ;  
  14.   
  15.     public   static   void  main(String[] args) {  
  16.         // ConnectionFactory :连接工厂,JMS 用它创建连接   
  17.         ConnectionFactory connectionFactory;  
  18.         // Connection :JMS 客户端到JMS Provider 的连接   
  19.         Connection connection = null ;  
  20.         // Session: 一个发送或接收消息的线程   
  21.         Session session;  
  22.         // Destination :消息的目的地;消息发送给谁.   
  23.         Destination destination;  
  24.         // MessageProducer:消息发送者   
  25.         MessageProducer producer;  
  26.         // TextMessage message;   
  27.         // 构造ConnectionFactory实例对象,此处采用ActiveMq的实现jar   
  28.   
  29.         connectionFactory = new  ActiveMQConnectionFactory(  
  30.                 ActiveMQConnection.DEFAULT_USER,  
  31.                 ActiveMQConnection.DEFAULT_PASSWORD,  
  32.                 "tcp://localhost:61616" );  
  33.         try  {  
  34.             // 构造从工厂得到连接对象   
  35.             connection = connectionFactory.createConnection();  
  36.             // 启动   
  37.             connection.start();  
  38.             // 获取操作连接   
  39.             session = connection.createSession(Boolean.TRUE,  
  40.                     Session.AUTO_ACKNOWLEDGE);  
  41.             // 获取session注意参数值xingbo.xu-queue是一个服务器的queue,
  42.                  须在在ActiveMq的console配置   
  43.             destination = session.createQueue("test-queue" );  
  44.             // 得到消息生成者【发送者】   
  45.             producer = session.createProducer(destination);  
  46.             // 设置不持久化,可以更改   
  47.             producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);  
  48.             // 构造消息   
  49.             sendMessage(session, producer);  
  50.             session.commit();  
  51.   
  52.         } catch  (Exception e) {  
  53.             e.printStackTrace();  
  54.         } finally  {  
  55.             try  {  
  56.                 if  ( null  != connection)  
  57.                     connection.close();  
  58.             } catch  (Throwable ignore) {  
  59.             }  
  60.         }  
  61.   
  62.     }  
  63.     public   static   void  sendMessage(Session session, MessageProducer producer)  
  64.             throws  Exception {  
  65.         for  ( int  i =  1 ; i <=SEND_NUMBER; i++) {  
  66.             TextMessage message = session  
  67.                     .createTextMessage("ActiveMq 发送的消息"  + i);  
  68.             // 发送消息到目的地方   
  69.             System.out.println("发送消息:"  + i);  
  70.             producer.send(message);  
  71.         }  
  72.     }  
  73. }  
 

       3:接收端,receive

 

Java代码 
  1. import  javax.jms.Connection;  
  2. import  javax.jms.ConnectionFactory;  
  3. import  javax.jms.Destination;  
  4. import  javax.jms.JMSException;  
  5. import  javax.jms.Message;  
  6. import  javax.jms.MessageConsumer;  
  7. import  javax.jms.MessageListener;  
  8. import  javax.jms.Session;  
  9. import  javax.jms.TextMessage;  
  10.   
  11. import  org.apache.activemq.ActiveMQConnection;  
  12. import  org.apache.activemq.ActiveMQConnectionFactory;  
  13.   
  14. public   class  Receiver {  
  15.     public   static   void  main(String[] args) {  
  16.   
  17.         // ConnectionFactory :连接工厂,JMS 用它创建连接   
  18.         ConnectionFactory connectionFactory;  
  19.         // Connection :JMS 客户端到JMS Provider 的连接   
  20.         Connection connection = null ;  
  21.         // Session: 一个发送或接收消息的线程   
  22.         Session session;  
  23.         // Destination :消息的目的地;消息发送给谁.   
  24.         Destination destination;  
  25.         // 消费者,消息接收者   
  26.         MessageConsumer consumer;  
  27.   
  28.         connectionFactory = new  ActiveMQConnectionFactory(  
  29.                 ActiveMQConnection.DEFAULT_USER,  
  30.                 ActiveMQConnection.DEFAULT_PASSWORD, "tcp://localhost:61616" );  
  31.         try  {  
  32.             // 构造从工厂得到连接对象   
  33.             connection = connectionFactory.createConnection();  
  34.             // 启动   
  35.             connection.start();  
  36.             // 获取操作连接   
  37.             session = connection.createSession(Boolean.FALSE,  
  38.                     Session.AUTO_ACKNOWLEDGE);  
  39.             //test-queue跟sender的保持一致,一个创建一个来接收   
  40.             destination = session.createQueue("test-queue" );  
  41.             consumer = session.createConsumer(destination);  
  42.             consumer.setMessageListener(new  MessageListener() {  
  43.                 public   void  onMessage(Message arg0) {  
  44.                     System.out.println("==================" );  
  45.                     try  {  
  46.                         System.out.println("RECEIVE1第一个获得者:"   
  47.                                 + ((TextMessage) arg0).getText());  
  48.                     } catch  (JMSException e) {  
  49.                         // TODO Auto-generated catch block   
  50.                         e.printStackTrace();  
  51.                     }  
  52.   
  53.                 }  
  54.             });  
  55.   
  56.             MessageConsumer consumer1 = session.createConsumer(destination);  
  57.             consumer1.setMessageListener(new  MessageListener() {  
  58.                 public   void  onMessage(Message arg0) {  
  59.                     System.out.println("+++++++++++++++++++" );  
  60.                     try  {  
  61.                         System.out.println("RECEIVE1第二个获得者:"   
  62.                                 + ((TextMessage) arg0).getText());  
  63.                     } catch  (JMSException e) {  
  64.                         // TODO Auto-generated catch block   
  65.                         e.printStackTrace();  
  66.                     }  
  67.   
  68.                 }  
  69.             });  
  70.         } catch  (Exception e) {  
  71.             e.printStackTrace();  
  72.         }  
  73.         //在eclipse里运行的时候,这里不要关闭,这样就可以一直等待服务器发送了,不然就直接结束了。   
  74.         // } finally {   
  75.         // try {   
  76.         // if (null != connection)   
  77.         // connection.close();   
  78.         // } catch (Throwable ignore) {   
  79.         // }   
  80.         // }   
  81.   
  82.     }  
  83. }  
 

 

        4:发送端,sender  上面的是用Queue的方式来创建,下面再用topic的方式实现同样的功能。

 

Java代码 
  1. import  javax.jms.Connection;  
  2. import  javax.jms.JMSException;  
  3. import  javax.jms.Message;  
  4. import  javax.jms.MessageConsumer;  
  5. import  javax.jms.MessageListener;  
  6. import  javax.jms.MessageProducer;  
  7. import  javax.jms.Session;  
  8. import  javax.jms.TextMessage;  
  9. import  javax.jms.Topic;  
  10.   
  11. import  org.apache.activemq.ActiveMQConnectionFactory;  
  12. import  org.apache.activemq.command.ActiveMQTopic;  
  13.   
  14. public   class  TopicTest {  
  15. 分享到:
    评论

相关推荐

Global site tag (gtag.js) - Google Analytics