ActiveMQ - Subscriber

• 

  简述

  现在让我们创建一个订阅者应用程序，它将接收来自 ActiveMQ 主题的消息。
• 

  创建项目

  使用eclipse，选择File → New→Maven Project. 勾选Create a simple project(skip archetype selection) 然后单击下一步。

  输入详细信息，如下所示 -

  • groupId − com.jc2182
  • artifactId − subscriber
  • version − 0.0.1-SNAPSHOT
  • name − ActiveMQ Subscriber

  单击完成按钮，将创建一个新项目。
• 

  pom.xml

  现在更新 pom.xml 的内容以包含 ActiveMQ 的依赖项。

  
  <project xmlns="http://maven.apache.org/POM/4.0.0" 
     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"    
     xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 
     https://maven.apache.org/xsd/maven-4.0.0.xsd">
     
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.jc2182.activemq</groupId>
     <artifactId>subscriber</artifactId>
     <version>0.0.1-SNAPSHOT</version>
     <name>ActiveMQ Subscriber</name>
     <dependencies>
        <dependency>
           <groupId>org.apache.geronimo.specs</groupId>
           <artifactId>geronimo-jms_1.1_spec</artifactId>
           <version>1.1</version>
        </dependency>
        <dependency>
           <groupId>org.apache.qpid</groupId>
           <artifactId>qpid-jms-client</artifactId>
           <version>0.40.0</version>
        </dependency>
     </dependencies>
     <build>
        <plugins>
           <plugin>
              <groupId>org.apache.maven.plugins</groupId>
              <artifactId>maven-compiler-plugin</artifactId>
              <version>2.1</version>
              <configuration>
                 <source>1.6</source>
                 <target>1.6</target>
              </configuration>
           </plugin>
           <plugin>
              <groupId>org.fusesource.mvnplugins</groupId>
              <artifactId>maven-uberize-plugin</artifactId>
              <version>1.14</version>
              <executions>
                 <execution>
                    <phase>package</phase>
                    <goals><goal>uberize</goal></goals>
                 </execution>
              </executions>
           </plugin>
        </plugins>
     </build>  
  </project>
  

  现在创建一个订阅者类，它将接收来自 ActiveMQ 队列的消息。

  
  package com.jc2182.activemq;
  
  import java.io.Console;
  
  import javax.jms.Connection;
  import javax.jms.Destination;
  import javax.jms.Message;
  import javax.jms.MessageConsumer;
  import javax.jms.MessageProducer;
  import javax.jms.Session;
  import javax.jms.TextMessage;
  
  import org.apache.qpid.jms.JmsConnectionFactory;
  
  public class Subscriber {
     public static void main(String[] args) throws Exception {
        // Create a connection to ActiveMQ JMS broker using AMQP protocol
        JmsConnectionFactory factory = new JmsConnectionFactory("amqp://localhost:5672");
        Connection connection = factory.createConnection("admin", "password");
        connection.start();
  
        // Create a session
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  
        // Create a topic
        Destination destination = session.createTopic("MyFirstTopic");
  
        // Create a subscriber specific to topic
        MessageConsumer subscriber = session.createConsumer(destination);
  
        Console c = System.console();
        String response;
        do {        
           // Receive the message
           Message msg = subscriber.receive();
           response = ((TextMessage) msg).getText();
  
           System.out.println("Received = "+response);
  
        } while (!response.equalsIgnoreCase("Quit"));
  
        // Close the connection
        connection.close();
     }
  }
  

  订阅者类创建一个连接，启动会话，创建一个消费者，然后从主题接收消息（如果有的话）。如果主题包含 quit as message 则应用程序终止，否则它将继续轮询队列以获取消息。

  我们将多次运行此应用程序以在ActiveMQ - 测试应用程序一章中创建多个订阅者。