Apache ActiveMQ - 订阅者应用程序
现在让我们创建一个订阅者应用程序,它将接收来自 ActiveMQ 主题的消息。
创建项目
使用 eclipse,选择 File → New → Maven Project。 勾选 Create a simple project(skip archetype selection) 并单击下一步。
输入详细信息,如下图 −
groupId − com.tutorialspoint
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.tutorialspoint.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.tutorialspoint.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(); } }
订阅者类创建一个连接,启动会话,创建一个消费者,然后从主题接收消息(如果有)。 如果主题包含作为消息的退出,则应用程序终止,否则它将继续轮询队列以获取消息。
我们将多次运行此应用程序以在 ActiveMQ - 测试应用程序章节中创建多个订阅者。