ActiveMQ - 消费者

  • 简述

    现在让我们创建一个消费者应用程序,它将接收来自 ActiveMQ 队列的消息。
  • 创建项目

    使用eclipse,选择FileNewMaven Project. 勾选Create a simple project(skip archetype selection) 然后单击下一步。
    输入详细信息,如下所示 -
    • groupId- com.jc2182
    • artifactId− consumer
    • version− 0.0.1-SNAPSHOT
    • name− ActiveMQ Consumer
    单击完成按钮,将创建一个新项目。
  • 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>consumer</artifactId>
       <version>0.0.1-SNAPSHOT</version>
       <name>ActiveMQ Consumer</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>
    
    现在创建一个 Consumer 类,它将接收来自 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 Consumer {
       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 queue
          Destination destination = session.createQueue("MyFirstQueue");
    
          // Create a consumer specific to queue
          MessageConsumer consumer = session.createConsumer(destination);
    
          Console c = System.console();
          String response;
          do {        
             // Receive the message
             Message msg = consumer.receive();
             response = ((TextMessage) msg).getText();
    
             System.out.println("Received = "+response);
    
          } while (!response.equalsIgnoreCase("Quit"));
    
          // Close the connection
          connection.close();
       }
    }
    
    消费者类创建一个连接,启动会话,创建一个消费者,然后从队列中接收消息(如果有的话)。如果队列包含作为消息退出,则应用程序终止,否则它将继续轮询队列以获取消息。
    我们将在ActiveMQ - 测试应用程序一章中运行这个应用程序。