activemq

ActiveMQ实战之 Topic发布订阅消息

消息发布者

view plaincopy to clipboardprint?

    package com.googlecode.garbagecan.jmsstudy.activemq.topic;  
      
    import org.apache.activemq.ActiveMQConnectionFactory;  
      
    import javax.jms.*;  
      
    public class TopicPublisher {  
        public static void main(String[] args) throws JMSException {  
            ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");  
            Connection connection = factory.createConnection();  
            connection.start();  
              
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);  
            Topic topic = session.createTopic("myTopic.messages");  
      
            MessageProducer producer = session.createProducer(topic);  
            producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);  
      
            while(true) {  
                TextMessage message = session.createTextMessage();  
                message.setText("message_" + System.currentTimeMillis());  
                producer.send(message);  
                System.out.println("Sent message: " + message.getText());  
      
                try {  
                    Thread.sleep(1000);  
                } catch (InterruptedException e) {  
                    e.printStackTrace();  
                }  
            }  
      
    //      session.close();  
    //      connection.stop();  
    //      connection.close();  
        }  
    }  

消息订阅者（消息消费者）

view plaincopy to clipboardprint?

    package com.googlecode.garbagecan.jmsstudy.activemq.topic;  
      
    import org.apache.activemq.ActiveMQConnectionFactory;  
      
    import javax.jms.*;  
      
    public class TopicSubscriber {  
        public static void main(String[] args) throws JMSException {  
            ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");  
            Connection connection = factory.createConnection();  
            connection.start();  
              
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);  
            Topic topic = session.createTopic("myTopic.messages");  
      
            MessageConsumer consumer = session.createConsumer(topic);  
            consumer.setMessageListener(new MessageListener() {  
                public void onMessage(Message message) {  
                    TextMessage tm = (TextMessage) message;  
                    try {  
                        System.out.println("Received message: " + tm.getText());  
                    } catch (JMSException e) {  
                        e.printStackTrace();  
                    }  
                }  
            });  
    //      session.close();  
    //      connection.stop();  
    //      connection.close();  
        }  
    }  

JMS简介：一种应用于异步消息传递的标准API，JMS也是应用于程序间通讯的。但是，JMS与其他机制有所不同，主要表现在系统间传递信息的方式，见PPT1-2。简介传送也是JMS的关键。当一个应用程序通过JMS向另一个应用程序发送消息时，两个程序之间并没有直接的连接。发送应用程序会将消息交给一个服务，由服务确保将消息投递给接收应用程序。在JMS中有两个主要的概念：消息中介(message broker)和消息目标(destination)。当应用程序发送消息时，会将消息交给一个消息中介。消息中介实际上就是JMS版本的邮局。消息中介可以确保消息被投递到指定的消息目标，同时可以释放发送者，使其能够进行其他的业务。 在JMS中，每条消息带有一个消息目标。消息目标就好像一个邮箱，可以将消息放入这个邮箱，直到有人将它们取出。消息目标只关心消息应该从哪里获得--而不是由谁来获得。在JMS中，有两种消息目标类型：队列和主题。分别应用于队列的点对点模型或应用于主题的发布--订阅模型。

    点对点消息传递模型：在点对点模型中，每个消息都有一个发送者和一个接收者，见PPT3.在JMS中如果有多个接收者监听队列，就没办法直到某条特定的消息会被哪个接收者处理。这种不确定性实际上有很多好处，因为它可以让程序之用为队列添加监听器就能增大消息处理的能力。

    发布--订阅消息传递模型：在此模型中，消息会被发送给一个主题。像使用队列一样，可以让多个接收者监听一个主题。但是，与队列不同的是消息不再被只投递给一个接收者，所有主题的订阅者都会收到消息，见PPT4。

JMS的优点：不用等待--当使用JMS发送消息时，客户端不必等待消息被处理，甚至是被投递。客户端只需要将消息发给消息中介，就可以确信消息会被投递到适当的目标。因为不必等待，客户端就可以执行其他的任务。由于这种方法可以大大地节省时间，客户端的性能能够极大地提高。面向消息--与RFC通信面向方法调用不同，使用JMS发送消息是以数据位中心的。这意味着客户端不用固定搭配特定的方法符号。任何队列或主题订阅者都可以处理由客户端发送来的消息。客户端不必了解服务的任何规范。位置独立--JMS客户端不必知道由谁来处理他们的消息，或者服务的位置在哪里。客户端值需要了解需要通过哪个队列或主题发送消息。因此，只要能够从队列或主题获取消息，JMS客户端就不用在意服务的位置在哪里。确保投送--当使用JMS发送消息时，客户端能够确保消息被投递。即使在消息发送时服务无法使用，消息也会被储存起来，直到服务重新可以使用为止。

在Spring中安装ActiveMQ：它是一个开源消息中介（Apache的一个子项目），也是应用JMS异步消息传递的很好选择。需要从www.activemq.org下载包。在bin目录中，可以找到一个用于启动ActiveMQ的脚本：Unix用户的activemq或者Windows用户的activemq.bat。运行这个脚本，等ActiveMq启动后，就可以使用它进行中介服务了。在所有的示例中，都需要通过JMS连接工厂通过消息中介发送消息。我们选择了ActiveMQ作为消息中介，因此必须配置JMS连接工厂，使它能够了解如何连接到ActiveMQ,ActiveMQConnectionFactory是连接ActiveMQ的JMS连接工厂，在Spring中配置如下：

<bean id ="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory" >

       <property  name="brokerURL" value="tcp://localhost:61616">

</bean>

    声明一个ActiveMQ队列：

    

 <bean id="rantzDestination" class="org.apache.activemq.command.ActiveMQQueue">

            <constructor-arg index="0" value="rantz.marketing.queue" />

     </bean>

    声明一个ActiveMQ主题：

    

 <bean id="rantzDestination" class="org.apache.activemq.command.ActiveMQTopic">

            <constructor-arg index="0" value="rantz.marketing.topic" />

     </bean>

协同使用JMS和Spring:JMS为Java爱好者提供了一个与消息中介进行交互，以及发送和接收消息的标准API，而且每一个消息中介的实现都会支持JMS。因此你不必由于消息中介的不同而学习不同的消息传递的API。虽然JMS为所有的消息中介提供了统一接口，但这种接口用起来不是十分便利（见PPT5-6）

    使用JMS模板：JmsTemplate是Spring消除冗长和重复JMS代码的解决方案。JmsTemplate可以创建连接，获取会话，以及发送和接收消息。它使你可以专注于构建要发送的消息或者处理收到的消息。另外，JmsTemplate可以处理任何被抛出的JMSException。如果JMSException在JmsTemplate工作中被抛出，JmsTemplate将捕获这个异常，并且用一个未受查的JmsException的子类再次抛出它。见PPT7

    置入JmsTemplate模板：

    <bean id ="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">

       <property name="connectionFactory" ref="connectionFactory">

    </bean>

        发送消息：在RoadRantz端，我们使用JmsTemplate向RoandRantz市场发送驾驶员信息。RantzMarketingGatewayImpl是RoadRantz与市场系统进行交互的类。见PPT8，因此，当我们在Spring中配置RantzMarKetingGatewayImpl类时，必须置入jmsTemplate和rantzDestination Bean的参考：

          <bean id="marketingGateway"  class="com.roadrantz.marketing.RantzMarketingGatewayImpl">

                <property name="jmsTemplate" ref ="jmsTemplate"  />

                <property name="destination" ref="rantzDestination"  />

         </bean>

        设置默认目标：如果每次发送消息都指定一个目标，不如为JmsTemplate置入一个默认目标：

      

 <bean id ="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">

           <property name="connectionFactory" ref="connectionFactory" />

          <property name="defaultDestination" ref="rantzDestination" />

        </bean>

        现在，调用JmsTemplate的send()方法就可以去掉第一个参数了。这种send()方法的形式只带有一个MessageCreator。因为没有指定目标，JmsTemplate会假设你要将消息发送给默认目标。所以不必再为RantzMarketingGatewayImpl注入目标了。它的声明很简单：

         

<bean id="marketingGateway"  class="com.roadrantz.marketing.RantzMarketingGatewayImpl">

                <property name="jmsTemplate" ref ="jmsTemplate"  />

         </bean>

        消费消息：使用JmsTemplate接收消息十分简单。只需要调用JmsTemplate的receive()方法，如PPT9. 默认情况下，对receive()方法的调用会造成阻塞，直到消息到达目标--如果必要，会永远等待下去。为了避免对消息的内部等待，可以通过配置JmsTemplate时，通过设置receiveTimeout属性来指定接收时间。下面配置的接收超时时间为一分钟（60000毫秒）。

       

<bean id ="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">

           <property name="connectionFactory" ref="connectionFactory" />

          <property name="defaultDestination" ref="rantzDestination" />

          <property name="receiveTimeout" value="60000" />

        </bean>

        在PPT9中，receive()方法会从默认目标接收消息。如果你希望指定一个目标，可以为其传递一个目标：

     

  MapMessage message = (MapMessage) jmsTemplate.receive(destination);

        此外，还可以通过名称指定目标，并且让Spring的目标解析器自动解析这个目标：

      

  MapMessage message = (MapMessage) jmsTemplate.receive("rantz.marketing.queue");

        转换消息：为了简化示例，没有在发送和接收消息的代码中添加处理消息转换的代码。但是，如果在程序中的多个位置都需要发送或接收相同的消息，可能就需要通过消息转换器来避免不必要的映射代码。尽管将消息转换代码抽象成自己的工具类并不是很困难，但是你仍然需要明确的调用工具类来进行转换。幸运的是，Spring通过MessageConverter接口提供了对消息转换的支持：

          

 public interface MessageConverter {

                public Message toMessage(Object  object, Session session);

                public Object fromMessage (Message message) ;

         } 

           见PPT10，展示了一个MessageConverter接口的实现MotoristMessageConverter,可以将Motorist对象转换为消息，也可以将消息转换为SpammedMotorist对象。

            发送和接收被转换的消息：我们可以在发送消息前不用明确的调用toMessage()方法，只需要调用JmsTemplate的convertAndSend()方法。因此，PPT8中的sendMotoristInfo()方法将变得更加简单：
                

  public void sendMotoristInfo (final Motorist motorist){
                       jmsTemplate.converAndSend(motorist);  }  

    消息被发向JmsTemplate的默认目标(假设已经指定了一个默认目标)。不过我们也可以在调用convertAndSend()方法时指定一个特定的目标：            

jmsTemplate.connverAndSend(destination,motorist);  

另外，我们也可以通过名称指定目标：                    

jmsTemplate.convertAndSend("rantz.marketing.queue",motorist); 

接收端：         public SpammedMotorist receiveSpammedMotorist() {    return (SpammedMotorist)

 jmsTemplate.receiveAndConvert();        }  

除非有特殊的指定，recieveAndConvert()会从默认目标接收消息。不过，我们也可以通过为receiveAndConvert()方法传递一个参数来选择一个目标：          return (SpammedMotorist) jmsTemplate.receiveAndConvert(destination);    或者使用目标的名称： return (SpammedMotorist) jmsTemplate.receiveAndConvert("rantz.marketing.queue");

            置入消息转化器：

  <bean id="motoristConverter" class="com.roadrantz.marketing.MotoristMessageConverter" />

            最后，JmsTemplate需要了解这个消息转换器。为了提供消息转换器，我们会将motoristConverter Bean置入到JmsTemplate的messageConverter属性中：

           

<bean id ="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">

               <property name="connectionFactory" ref="connectionFactory" />

              <property name="defaultDestination" ref="rantzDestination" />

              <property name="messageConverter" ref="motoristConverter" />

            </bean>

    将Spring的网关支持类应用于JMS: Spring通过提供JdbcDaoSupport使JdbcTemplate的应用比原先变得更简单一些。与此类似，Spring也提供了一个用于JMS网关类的基类JmsGatewaySupport(见PPT11)。 那么JmsGatewaySupport从哪里得到JmsTemplate呢？你可以直接在jmsTemplate属性中注入JmsTemplate，就像处理常规的RantzMarketingGatewayImpl一样。或者，可以为connectionFactory属性置入连接工厂，来满足对JmsTemplate Bean的所有需求：

    

<bean id = "marketingGateway" class="com.roadrantz.marketing.RantzMarketingGatewayImple">

        <property name="connectionFactory " ref="connectionFactory" />

    </bean>

      当这种方法配置时，JmsGateSupport将基于被注入的连接工厂自动创建一个JmsTemplate对象。因此不必在Spring中声明JmsTemplate Bean。 在直接将连接工厂置入到网关之前，你应该了解这种方法有两个缺点：

    1、只能在JmsTemplate上指定默认目标。如果JmsGateSupport创建了自己的JmsTemplate，就没有机会再指定默认的目标了。你必须在调用send()或receive()时明确地选择一个目标。

    2、只能将消息转换器置入到JmsTemplate中。如果JmsGatewaySupport创建了自己JmsTemplate，将不能使用消息转换器。因此，必须明确地在网关代码中处理消息的转换。

创建消息驱动POJO：EJB2.0规范的其中一个重要内容是包含了消息驱动Bean(MDB)。MDB是可以异步处理消息的EJB。换句话说，MDB会将JMS目标中的消息作为事件来响应。这与同步消息接收者在消息可用前进行阻塞正好相反。MDB是EJB的一个两点。在EJB3的规范中，MDB被简化了，使其更像POJO。不再需要实现MessageDrivenBean接口，只需要实现常规的java.jms.MessageListener接口并用@MessageDriven注释MDB。Spring2.0通过自己的消息驱动Bean形式来满足消息的异步消费需求。

    创建消息监听器：MarketingMdp不必实现MessageDrivenBean接口的世界是多么简单，见PPT12。MarketingMdp本身并不做些什么。它有一个实际处理消息的onMessage()方法。不过要先在Spring中配置一下：

  

 <bean id="rantzMdp" class="com.roadrantz.marketing.MarketingMdp"  />

    EJB3.0MDB会使用@MessageDriven注释通知容器这是一个MDB。但是，在Spring中，我们会通过将其注入到一个消息监听容器来指示这个Bean是一个MDP。

    包含消息监听器：消息监听器容器是一个用于查看JMS目标，等待消息到达的特殊Bean。一旦消息到达，它就可以获取到消息，并通过onMessage()方法将消息传递给一个MessageListener的实现。因为MarketingMdp类实现了MessageListener接口，所以消息监听器容器就准备完毕了。见PPT13. 顾名思义，SimpleMessageListenerContainer是最简单的消息监听器容器，可以按下面的方法在Spring中进行配置：

   

<bean class="org.springframework.jms.listener.SimpleMessageListenerContainer">

         <property name="connectionFactory" ref="connectionFactory" />

         <property name="destination" ref="rantzDestination" />

         <property name="messageListener" ref="rantzMdp" />

    </bean> 

    对于messageListener属性，我们为其置入了对MDP实现的引用，这样，onMessage()方法将可以被用来接收消息。

    使用事务性的MDP:如果收到的一个消息在事务中，则应该使用DefaultMessageListenerContainer：

   

 <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">

         <property name="connectionFactory" ref="connectionFactory" />

         <property name="destination" ref="rantzDestination" />

         <property name="messageListener" ref="rantzMdp" />

         <property name="transactionManager" ref="jmsTransactionManager" />

    </bean> 

如果事务性需求比较简单，JmsTransactionManager将按如下方法配置：

   

 <bean id="jmsTransactionManager" class="org.springframework.jms.connection.JmsTransactionManager">

         <property name="connectionFactory" ref="connectionFactory" />

    </bean> 

需要提醒的是，tansactionManger属性是可选的。如果不注入事务管理器，MDP就不是事务性的。

    编写纯POJO MDP：如果消息监听器容器的messageListener属性被注入了MessageListener的实现，它就能够知道在消息到达时应该调用onMessage()方法。幸运的是，Spring提供了一个替代的MessageListenerAdapter。它是一个MessageListener，可以委派Bean和你选择的方法，见PPT14. 如果不将自己的MessageListener的实现注入到消息监听器容器中，你可以置入到MessageListenerAdapter中：

  

 <bean class="org.springframework.jms.listener.SimpleMessageListenerContainer">

         <property name="connectionFactory" ref="connectionFactory" />

         <property name="destination" ref="rantzDestination" />

         <property name="messageListener" ref ="purePojoMdp" />

    </bean> 

因为配置了purePojoMdp Bean，所以它是一个MessageListenerAdapter:

   

 <bean id="purePojoMdp" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">

        <property name="delegate" ref="rantzMdp" />

        <property name="defaultListenerMethod"  value="processMotoristInfo" />

    </bean> 默认情况下，MessageListenerAdapter在消息到达时会调用handleMessage()方法。但是，我们希望MarketingMdp Bean可以通过 processMotoristInfo()方法处理消息，因此将defaultListenerMethod设置为processMotoristInfo.    因为选择了一个特定的被调用方法，所以不需要实现MessageListener或onMessage()方法。因此 MarketingMdp现在将被简化为PPT15。尽管它是一个POJO，对MapMessage的依赖造成了MarketingMdp与JMS的不必要耦合，另外，MapMessage的getString方法还会抛出必须被处理的JMSException。理想情况下，MarketingMdp不应该依赖任何特定框架的类型。 当MessageListenerAdapter接收消息时，它会考虑消息的类型和defaultListenerMethod的值，并且尝试着查找用来调用的监听器方法符号。PPT16描述了MessageListenerAdapter是如何将JMS消息映射到监听器方法参数的。

    转换MDP消息：在最新版中，processMotoristInfo()带有的是Map，并且在处理前需要将Map转换为SpammedMotorist。如果在消息到达时，能够直接给processMotoristInfo()方法传递可以处理的SpammedMotorist对象岂不是更好么？Spring消息转换器可以执行消息和特定域Java类型之间的相互转换工作。在PPT10中，已经有一个消息转换器。我们需要做的就是让MessageListenerAdapter能够感知这个消息转换器。MessageListenerAdapter的messageConverter属性可以完成这项工作：

  

 <bean id="purePojoMdp" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">

        <property name="delegate" ref="rantzMdp" />

        <property name="defaultListenerMethod"  value="processMotoristInfo" />

        <propert name="messageConverter" ref="motoristConverter" />

    </bean> 

现在，可以编写最终版的MarketingMdp了。

使用基于消息的RPC：RPC编程模型可以使与远程服务的交互就如同在调用本地对象的方法。是否有某种方法既具有RPC编程模型的简单性，又可以利用异步消息传递的优点呢？

    引入Lingo： 它是一种基于Spring的远程调用方法，它在RPC和异步消息传递之间架起了一座桥梁。与使用其他的Spring远程调用方法相同，Lingo提供了一个服务输出器，可以将Bean的功能输出为Lingo服务和客户端代理。 Lingo远程调用会通过JMS队列或主题来承载信息。 尽管Lingo是基于Spring远程调用的，但是它并不是Spring框架的一部分。你可以从Lingo的主页下载Lingo(http://lingo.codehaus.org/Download).

    输出服务：在服务端，Lingo提供了一个服务输出器JmsServiceExporter。被Lingo输出的服务是通过JMS来提供服务的，而不是用于直接的RPC访问，见PPT17。下面的XML配置了一个JmsServiceExporter，能够将rantzMdp Bean输出为RPC-over-JMS服务：

   

<bean id="server" class="org.logicblaze.lingo.jms.JmsServiceExporter">

         <property name="connectionFactory" ref="connectionFactory" />

         <property name="destination" ref="destination" />

         <property  name="service" ref="rantzMdp" />

         <property name="serviceInterface" value="com.roadrantz.marketing.MarketingService"  />

    </bean>

      service属性被置入了对rantzMdp Bean的引用，即MarketingMdp。最后，应该将定义了服务的接口的类名配置到serviceInterface属性中。我们声明的被输出服务带有MarketingService接口，定义如下：

    

 public interface MarketingService {

        void processMotoristInfo (SpammedMotorist motorist );

    }   

因为我们定义的服务带有MarketingService接口，这就意味着我们应该对MarketingMdp类进行一些小的修改，以便它可以实现MarketingService接口：

    

 public class MarketingMdp implements MarketingService {

         public void processMotoristInfo(SpammedMotorist motorist) {

             ...

        }

    }

    这就是使用Lingo输出服务的全过程。一旦应用程序启动，JmsServiceExporter将会起作用，就可以开始使用它了。现在，让我们再到客户端看看RoadRantz应用程序是如何调用这个被输出的市场服务的。

    代理JMS：在RoadRantz应用程序中，每次用户注册并选择愿意接收特定提供商信息时都需要调用processMotoristInfo()方法，因此，我们必须以某种方法将Lingo输出服务的引用置入到RoadRantz应用程序中。

        置入JmsProxyFactoryBean:lingo提供了JmsProxyFactoryBean，这是一个代理工厂Bean，可以生成远程Lingo输出服务的代理。如PPT18所示，通过JmsProxyFactoryBean代理的服务是通过JMS目标(队列或主题)访问的，而不是通过TCP/IP。下面声明配置了一个JmsProxyFactoryBean：

        <bean id = "marketing" class="org.logicblaze.lingo.jms.JmsProxyFactoryBean">

             <property name="connectionFactory" ref="connectionFactory" />

             <property name="destination" ref="destination" />

             <property name="serviceInterface" value="com.roadrantz.marketing.MarketingService" />

        </bean>

serviceInterface属性指定了代理需要实现的Java接口。通过这个接口，RoadRantz可以调用processMotoristInfo()方法。 对于配置JmsProxyFactoryBean，最需要注意的事情是不用配置服务的位置信息。这是因为服务的位置并不重要，只需要知道服务在哪里接收"邮件"。事实上，我们还可以使用远程服务的多个实例。如果想建立一个高可靠性的市场服务，可以启动两个或多个实例，让它们都监听相同的目标。每一个实例都能够处理一个请求。其间，客户端并不知道有一个服务池在等待请求进行响应。

        进行调用：在JmsProxyFactoryBean被置入后，就可以开始调用远程服务了。我们需要做的是将其置入到RantServiceImpl:

     

  <bean id ="rantService" class="com.roadrantz.service.RantServiceImpl">

           <property name="rantDao" ref="rantDao" />

           <property name="marketingService" ref="marketing" />

        </bean>

  接着，我们可以使用它在addMotorist()方法中向市场服务发送SpammedMotorist对象。PPT19展示了为了调用远程市场服务对RantServiceImpl进行的相关修改。 正如你看到的，调用Lingo输出的服务与调用RMI服务、Web服务或在同一个进程中调用其他Bean上的方法是相同的。PPT19中没有任何关于JMS的内容。唯一不同的地方在Spring的配置。通过这种方法，只需要简单的更改Spring的配置就能够在JMS和其他通信机制直接进行切换。