hk2000c技术专栏

技术源于哲学，哲学来源于生活 关心生活，关注健康，关心他人

  BlogJava :: 首页 :: 新随笔 :: 联系 :: 聚合  :: 管理 ::
  111 随笔 :: 1 文章 :: 28 评论 :: 0 Trackbacks
ActiveMQ 实践之路(二) 使用Queue或者Topic发送/接受消息

本篇主要讲解在未使用其他框架(Spring)整合情况下,独立基于ActiveMQ,使用JMS规范进行消息通信。
    
     一.JMS回顾
       因为ActiveMQ是一个JMS Provider的实现,因此在开始实作前,有必要复习下JMS的基础知识
    Java Message Service (JMS)是sun提出来的为J2EE提供企业消息处理的一套规范,JMS目前有2套规范还在使用JMS 1.0.2b和1.1. 1.1已经成为主流的JMS Provider事实上的标准了.
      *1.1主要在session上面有一些重要改变,比如支持建立同一session上的transaction,让他支持同时发送P2P(Queue)消息和接受
Topic消息。
      
       在JMS中间主要定义了2种消息模式Point-to-Point (点对点),Publich/Subscribe Model (发布/订阅者)，
    其中在Publich/Subscribe 模式下又有Nondurable subscription和durable subscription (持久化订阅)2种消息处理方式。
    
     下面是JMS规范基本的接口和实现
     JMS Common Interfacse PTP-Specific Interface   Pub/Sub-specific interfaces
     ConnectionFactory     QueueConnectionFactory   TopicConnectionFactory
     Connection            QueueConnection          TopicConnection
     Destination           Queue                    Topic
     Session               QueueSession             TopiSession
     MessageProducer       QueueSender              TopicPublisher
     MessageConsumer       QueueReceiver/QueueBrwer TopicSubscriber


     二.使用Queue

         下面以ActiveMQ example的代码为主进行说明
        
        1. 使用ActiveMQ的Connection，ConnectionFactory 建立连接,注意这里没有用到pool
       

java 代码
  1. import org.apache.activemq.ActiveMQConnection   
  2. import org.apache.activemq.ActiveMQConnectionFactory   

        //建立Connection

java 代码
  1. protected Connection createConnection() throws JMSException, Exception {   
  2.      ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, pwd, url);   
  3.      Connection connection = connectionFactory.createConnection();   
  4.      if (durable && clientID!=null) {   
  5.          connection.setClientID(clientID);   
  6.      }   
  7.      connection.start();   
  8.      return connection;   
  9.     }  

        //建立Session
  

java 代码
  1. protected Session createSession(Connection connection) throws Exception {   
  2.          Session session = connection.createSession(transacted, ackMode);   
  3.          return session;   
  4.         }   

        2。发送消息的代码
 //建立QueueSession
 

java 代码
  1. protected MessageProducer createProducer(Session session) throws JMSException {   
  2.         Destincation destination = session.createQueue("queue.hello");   
  3.         MessageProducer producer = session.createProducer(destination);   
  4.         producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);   
  5.            
  6.         if( timeToLive!=0 )   
  7.             producer.setTimeToLive(timeToLive);   
  8.         return producer;   
  9.         }   

         //使用Producer发送消息到Queue
    

java 代码
  1. producer.send(message);   

       
        3。接受消息,在JMS规范里面,你可以使用
  

java 代码
  1. QueueReceiver/QueueBrowser直接接受消息,但是更多的情况下我们采用消息通知方式,即实现MessageListener接口   
  2.  public void onMessage(Message message) {   
  3.  //process message   
  4.  }   
  5.           
  6.  //set MessageListner ,receive message   
  7.  Destincation destination = session.createQueue("queue.hello");   
  8.  consumer = session.createConsumer(destination);   
  9.  consumer.setMessageListener(this);   

       
        以上就是使用jms queue发送接受消息的基本方式

 
     三 Topic

        1. 建立连接
   

java 代码
  1. protected Connection createConnection() throws JMSException, Exception {      
  2.         ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, pwd, url);      
  3.         Connection connection = connectionFactory.createConnection();      
  4.         //如果你要使用DurableSubScription 方式,你必须为connection设置一个ClientID      
  5.         if (durable && clientID!=null) {      
  6.             connection.setClientID(clientID);      
  7.         }      
  8.         connection.start();      
  9.         return connection;      
  10.        }      

       2. 建立Session

java 代码
  1. protected Session createSession(Connection connection) throws Exception {      
  2.         Session session = connection.createSession(transacted, ackMode);      
  3.         return session;      
  4.         }    

 3.创建Producer 发送消息到Topic   
       

java 代码
  1. //create topic on  session   
  2.        topic = session.createTopic("topic.hello");   
  3.  producer = session.createProducer(topic);   
  4.        //send message    
  5.        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);   
  6.  producer.send(message);   


 4.创建Consumer接受消息(基本上和Queue相同)

java 代码
  1. Destincation destination  = session.createTopic("topic.hello");      
  2. MessageConsumer consumer = session.createConsumer(destination);      
  3. consumer.setMessageListener(this);      
  4.            
  5.      //如果你使用的是Durable Subscription方式,你必须在建立connection的时候      
  6.      //设置ClientID,而且建立comsumer的时候使用createDurableSubscriber方法,为他指定一个consumerName。      
  7.  //connection.setClientID(clientId);      
  8.  //consumer = session.createDurableSubscriber((Topic) destination, consumerName);   

       
 四:连接ActiveMQ的方式
        ActiveMQConnectionFactory 提供了多种连接到Broker的方式activemq.apache.org/uri-protocols.html

 常见的有
 vm://host:port     //vm
 tcp://host:port    //tcp
 ssl://host:port    //SSL
 stomp://host:port  //stomp协议可以跨语言,目前有很多种stomp client 库(java,c#,c/c++,ruby,python...);
posted on 2007-11-16 17:05 hk2000c 阅读(8347) 评论(2)  编辑  收藏 所属分类: JMS

评论

# re: ActiveMQ 实践之路(二) 使用Queue或者Topic发送/接受消息 2009-06-03 16:43 蓝雨
写的太简单了， 初学者肯定看不懂！  回复  更多评论
  

# re: ActiveMQ 实践之路(二) 使用Queue或者Topic发送/接受消息 2009-10-21 09:08 READ
确实如楼上所说...  回复  更多评论
  


只有注册用户登录后才能发表评论。
网站导航: