Apache Activemq 简明教程
Apache ActiveMQ - Overview
What is ActiveMQ?
ActiveMQ 是使用 Java 编写的开源消息代理。它完全符合 JMS 1.1 标准。它是由 Apache 软件基金会开发和维护的,并在 Apache 许可证下获得许可。它为企业级消息传递应用程序提供高可用性、可伸缩性、可靠性、性能和安全性。
JMS 是一个允许开发基于消息的系统的规范。ActiveMQ 充当消息代理,位于应用程序之间,并允许它们以异步且可靠的方式进行通信。
Apache ActiveMQ - Environment Setup
本篇指导将引导你如何准备开发环境来开始使用 ActiveMQ 工作。它还将教你如何在设置 ActiveMQ 之前在计算机上设置 JDK、Maven 和 Eclipse −
Setup Java Development Kit (JDK)
你可以从 Oracle 的 Java 站点下载 SDK 的最新版本 − Java SE Downloads. 你在下载文件中可以找到安装 JDK 的说明,按照给定的说明进行安装和配置。最后设置 PATH 和 JAVA_HOME 环境变量以引用包含 java 和 javac 的目录,通常分别为 java_install_dir/bin 和 java_install_dir。
如果您运行的是 Windows 且在 C:\jdk-11.0.11 中安装了 JDK,那么您将不得不把以下代码行放入您的 C:\autoexec.bat 文件中。
set PATH=C:\jdk-11.0.11;%PATH%
set JAVA_HOME=C:\jdk-11.0.11
或者,在 Windows NT/2000/XP 中,你必须右键单击我的电脑,选择属性 → 高级 → 环境变量。然后,你将不得不更新 PATH 值并单击确定按钮。
在 Unix(Solaris、Linux 等)中,如果 SDK 安装在 /usr/local/jdk-11.0.11 且您使用 C shell,那么您将不得不把以下代码行放入您的 .cshrc 文件中。
setenv PATH /usr/local/jdk-11.0.11/bin:$PATH
setenv JAVA_HOME /usr/local/jdk-11.0.11
或者,如果你使用诸如 Borland JBuilder、Eclipse、IntelliJ IDEA 或 Sun ONE Studio 这样的集成开发环境 (IDE),则必须编译并运行一个简单程序来确认 IDE 知道你在何处安装了 Java。否则,你必须按照 IDE 文档中给出的内容执行正确的设置。
Setup Eclipse IDE
本教程中的所有示例都是使用 Eclipse IDE 编写的。因此,我们建议你应该在你机器上安装 Eclipse 的最新版本。
要安装 Eclipse IDE,请从 www.eclipse.org/downloads/ 下载最新的 Eclipse 二进制文件。下载安装文件后,将二进制分发解压缩到合适的目录中。例如,在 Windows 中的 C:\eclipse 或 Linux/Unix 中的 /usr/local/eclipse 中,最后适当设置 PATH 变量。
可以通过在 Windows 机器上执行以下命令启动 Eclipse,或者你只需双击 eclipse.exe
%C:\eclipse\eclipse.exe
可以通过在 Unix(Solaris、Linux 等)机器上执行以下命令启动 Eclipse −
$/usr/local/eclipse/eclipse
成功启动后,如果一切正常,它应该显示以下结果 −
Set Maven
在本教程中,我们使用 Maven 来运行和构建基于 Spring 的示例,以运行基于 ActiveMQ 的应用程序。按照 Maven - Environment Setup 安装 Maven。
Download ActiveMQ
你可以从其官方页面下载最新稳定版的 ActiveMQ。按照 Download ActivMQ 下载 ActiveMQ。我们用了 2022 年 2 月 15 日发布的 5.13.4 版。将存档内容解压缩到你选择的文件夹中。我们已经解压缩到 F:/ → Apache → apache-activemq-5.16.4 中。
Apache ActiveMQ - Features
ActiveMQ 旨在为企业级消息传递应用程序提供高可用性、可伸缩性、可靠性、性能和安全性。以下是 ActiveMQ 的一些显著特性。
-
JMS Compliant − ActiveMQ 完全符合 JMS 1.1 标准。JMS 规范为同步或异步消息传递、仅传递一次的消息传递、消息对订户的耐用性等提供了标准机制。
-
Connectivity Options − ActiveMQ 支持 HTTP/S、多播、SSL、Stomp、TCP、UDP、XMPP,从而为连接提供了广泛的选择,并允许各种系统使用选择的协议进行通信。
-
Pluggable Architecture − ActiveMQ 允许选择持久性机制,还提供自定义安全性的选项,用于根据应用程序的需要进行身份验证和授权。
-
Multi-Platform − ActiveMQ 为许多流行的语言(如 Java、C、C++、.NET、Perl、PHP、Python、Ruby 等)提供客户端 API。ActiveMQ 代理将在 JVM 中运行,但客户端可以使用任何受支持的语言编写。
-
Broker Cluster − ActiveMQ 允许为可伸缩性准备代理网络,并可以支持不同类型的拓扑。
-
Features Rich − ActiveMQ 为代理和客户端提供许多高级功能,并支持 Apache Camel。
-
Simple Administration Interface − ActiveMQ 管理控制台易于使用,但仍提供许多强大的管理功能。
Apache ActiveMQ - Running Broker Server
我们已下载 ActiveMQ - Environment Setup 中提到的最新版 ActiveMQ。现在,转到文件夹 F:/ → Apache → apache-activemq-5.16.4/bin 并输入以下命令。
Output
您将看到类似的输出,ActiveMQ 将开始运行。
Java Runtime: Oracle Corporation 11.0.11 C:\Program Files\Java\jdk-11.0.11
Heap sizes: current=1048576k free=1041918k max=1048576k
JVM args: -Dcom.sun.management.jmxremote -Xms1G -Xmx1G -Djava.util.logging.config.file=logging.properties -Djava.security.auth.login.config=F:\Apache\apache-activemq-5.16.4\bin\..\conf\login.config -Dactivemq.classpath=F:\Apache\apache-activemq-5.16.4\bin\..\conf;F:\Apache\apache-activemq-5.16.4\bin\../conf;F:\Apache\apache-activemq-5.16.4\bin\../conf; -Dactivemq.home=F:\Apache\apache-activemq-5.16.4\bin\.. -Dactivemq.base=F:\Apache\apache-activemq-5.16.4\bin\.. -Dactivemq.conf=F:\Apache\apache-activemq-5.16.4\bin\..\conf -Dactivemq.data=F:\Apache\apache-activemq-5.16.4\bin\..\data -Djava.io.tmpdir=F:\Apache\apache-activemq-5.16.4\bin\..\data\tmp
Extensions classpath:
[F:\Apache\apache-activemq-5.16.4\bin\..\lib,F:\Apache\apache-activemq-5.16.4\bin\..\lib\camel,F:\Apache\apache-activemq-5.16.4\bin\..\lib\optional,F:\Apache\apache-activemq-5.16.4\bin\..\lib\web,F:\Apache\apache-activemq-5.16.4\bin\..\lib\extra]
ACTIVEMQ_HOME: F:\Apache\apache-activemq-5.16.4\bin\..
ACTIVEMQ_BASE: F:\Apache\apache-activemq-5.16.4\bin\..
ACTIVEMQ_CONF: F:\Apache\apache-activemq-5.16.4\bin\..\conf
ACTIVEMQ_DATA: F:\Apache\apache-activemq-5.16.4\bin\..\data
Loading message broker from: xbean:activemq.xml
INFO | Refreshing org.apache.activemq.xbean.XBeanBrokerFactory$1@53fe15ff: startup date [Sat Feb 26 12:50:18 IST 2022]; root of context hierarchy
INFO | Using Persistence Adapter: KahaDBPersistenceAdapter[F:\Apache\apache-activemq-5.16.4\bin\..\data\kahadb]
INFO | PListStore:[F:\Apache\apache-activemq-5.16.4\bin\..\data\localhost\tmp_storage] started
INFO | Apache ActiveMQ 5.16.4 (localhost, ID:DESKTOP-86KD9FC-52669-1645860020983-0:1) is starting
INFO | Listening for connections at: tcp://DESKTOP-86KD9FC:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600
INFO | Connector openwire started
INFO | Listening for connections at: amqp://DESKTOP-86KD9FC:5672?maximumConnections=1000&wireFormat.maxFrameSize=104857600
INFO | Connector amqp started
INFO | Listening for connections at: stomp://DESKTOP-86KD9FC:61613?maximumConnections=1000&wireFormat.maxFrameSize=104857600
INFO | Connector stomp started
INFO | Listening for connections at: mqtt://DESKTOP-86KD9FC:1883?maximumConnections=1000&wireFormat.maxFrameSize=104857600
INFO | Connector mqtt started
INFO | Starting Jetty server
INFO | Creating Jetty connector
WARN | ServletContext@o.e.j.s.ServletContextHandler@4f966719{/,null,STARTING} has uncovered http methods for path: /
INFO | Listening for connections at ws://DESKTOP-86KD9FC:61614?maximumConnections=1000&wireFormat.maxFrameSize=104857600
INFO | Connector ws started
INFO | Apache ActiveMQ 5.16.4 (localhost, ID:DESKTOP-86KD9FC-52669-1645860020983-0:1) started
INFO | For help or more information please see: http://activemq.apache.org
INFO | ActiveMQ WebConsole available at http://127.0.0.1:8161/
INFO | ActiveMQ Jolokia REST API available at http://127.0.0.1:8161/api/jolokia/
Verification
现在,在您的浏览器中打开 http://127.0.0.1:8161/admin/ 。它会要求提供凭据。使用 admin/admin 作为用户名/密码,它将加载 ActiveMQ 管理控制台,您可以在其中查看队列、主题、连接等。
Apache ActiveMQ - Admin Console
ActiveMQ 服务器启动并运行后。你可以使用管理控制台管理队列、主题、订户、连接、网络等。
在浏览器中打开 http://127.0.0.1:8161/admin/ 。它将询问凭据。使用 admin/admin 作为用户名/密码,它将加载 ActiveMQ 管理控制台,你可以在其中查看队列、主题、连接等。
Apache ActiveMQ - Producer Application
现在,我们来创建一个生产者应用程序,它将消息发送至 ActiveMQ 队列。
Create Project
使用 Eclipse,选择 File → New * → *Maven Project 。勾选 Create a simple project(skip archetype selection) ,然后单击下一步。
按照以下所示输入详细信息 −
-
groupId − com.tutorialspoint
-
artifactId − producer
-
version − 0.0.1-SNAPSHOT
-
name − ActiveMQ Producer
单击“完成”按钮,将创建新项目。
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>producer</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>ActiveMQ Producer</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 java.util.Scanner;
import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.qpid.jms.JmsConnectionFactory;
public class Producer {
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 producer specific to queue
MessageProducer producer = session.createProducer(destination);
Scanner input = new Scanner(System.in);
String response;
do {
System.out.println("Enter message: ");
response = input.nextLine();
// Create a message object
TextMessage msg = session.createTextMessage(response);
// Send the message to the queue
producer.send(msg);
} while (!response.equalsIgnoreCase("Quit"));
input.close();
// Close the connection
connection.close();
}
}
生产者类将创建连接、启动会话、创建一个生产者,然后让用户输入消息。如果用户输入退出,则应用程序终止,否则将发送消息至队列。
我们将在 ActiveMQ - Test Application 章节中运行此应用程序。
Apache ActiveMQ - Consumer Application
现在,我们来创建一个使用者应用程序,它将从 ActiveMQ 队列中接收消息。
Create Project
使用 eclipse,选择 File → New → Maven Project 。勾选 创建简单项目(跳过原型选择),然后单击下一步。
按照以下所示输入详细信息 −
-
groupId − com.tutorialspoint
-
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.tutorialspoint.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>
现在,创建一个消费者类,它将从 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 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 - Test Application 章节中运行此应用程序。
Apache ActiveMQ - Test Application
Start ActiveMQ Server
现在,我们来启动 ActiveMQ 服务器。转到文件夹 F:/ → Apache → apache-activemq-5.16.4/bin ,并输入以下命令。
Output
您将看到类似的输出,ActiveMQ 将开始运行。
...
INFO | Apache ActiveMQ 5.16.4 (localhost, ID:DESKTOP-86KD9FC-52669-1645860020983-0:1) started
INFO | For help or more information please see: http://activemq.apache.org
INFO | ActiveMQ WebConsole available at http://127.0.0.1:8161/
INFO | ActiveMQ Jolokia REST API available at http://127.0.0.1:8161/api/jolokia/
Start the Producer Application
在 eclipse 中,右键单击 Producer.java 源码,然后选择“Run As”(作为)> “Java 应用程序”。生产者应用程序将开始运行,您将看到如下输出:
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
Enter message:
Start the Consumer Application
在 eclipse 中,右键单击 Consumer.java 源码,然后选择“Run As”(作为)> “Java 应用程序”。使用者应用程序将开始运行,您将看到如下输出:
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
Send Message
在生产者控制台窗口中,键入 Hi 并按 enter 按钮发送消息。
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
Enter message:
Hi
Receive Message
在使用者控制台窗口中确认消息已收到。
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
Received = Hi
发送退出作为消息以终止生产者和使用者控制台窗口会话。
Verification
现在在浏览器中打开 http://127.0.0.1:8161/admin/ 。它会询问凭据。使用 admin/admin 作为用户名/密码,它将加载 ActiveMQ 管理控制台,您可以在其中查看队列以查看状态。它将显示已入队和已发送的 2 条消息。
Apache ActiveMQ - Publisher Application
现在,让我们创建一个发布者应用程序,它将把消息发送到 ActiveMQ 队列。
Create Project
使用 Eclipse 选择 File → New * → *Maven Project 。勾选 创建简单项目(跳过原型选择),然后单击下一步。
输入详细信息,如下所示:
-
groupId − com.tutorialspoint
-
artifactId − publisher
-
version − 0.0.1-SNAPSHOT
-
name − ActiveMQ Publisher
单击“完成”按钮,将创建新项目。
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>publisher</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>ActiveMQ Publisher</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 java.util.Scanner;
import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.qpid.jms.JmsConnectionFactory;
public class Publisher {
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 publisher specific to topic
MessageProducer publisher = session.createProducer(destination);
Scanner input = new Scanner(System.in);
String response;
do {
System.out.println("Enter message: ");
response = input.nextLine();
// Create a message object
TextMessage msg = session.createTextMessage(response);
// Send the message to the topic
publisher.send(msg);
} while (!response.equalsIgnoreCase("Quit"));
input.close();
// Close the connection
connection.close();
}
}
生产者类创建一个连接,启动会话,创建一个生产者,然后要求用户输入消息。如果用户输入退出,则应用程序终止,否则它将把消息发送给该主题。
我们将在 ActiveMQ - Test Application 章节运行此应用程序。
Apache ActiveMQ - Subscriber Application
现在,让我们创建一个将从 ActiveMQ 主题接收消息的订阅者应用程序。
Create Project
使用 eclipse,选择 File → New → Maven Project 。勾选 创建简单项目(跳过原型选择),然后单击下一步。
按照以下所示输入详细信息 −
-
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 - Test Application 章节中多次运行此应用程序以创建多个订阅者。
Apache ActiveMQ - Test Application Topic
Start ActiveMQ Server
现在,我们来启动 ActiveMQ 服务器。转到文件夹 F:/ → Apache → apache-activemq-5.16.4/bin ,并输入以下命令。
Output
您将看到类似的输出,ActiveMQ 将开始运行。
...
INFO | Apache ActiveMQ 5.16.4 (localhost, ID:DESKTOP-86KD9FC-52669-1645860020983-0:1) started
INFO | For help or more information please see: http://activemq.apache.org
INFO | ActiveMQ WebConsole available at http://127.0.0.1:8161/
INFO | ActiveMQ Jolokia REST API available at http://127.0.0.1:8161/api/jolokia/
Start the Publisher Application
在 eclipse 中,鼠标右键单击 Publisher.java 源,然后选择以 Java 应用程序运行。Publisher 应用程序将开始运行,您会看到如下所示的输出:
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
Enter message:
Start the Subscriber Application
在 eclipse 中,鼠标右键单击 Subscriber.java 源,然后选择以 Java 应用程序运行。Subscriber 应用程序将开始运行,您会看到如下所示的输出:
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
Start another Subscriber Application
在 eclipse 中,再次鼠标右键单击 Subscriber.java 源,然后选择以 Java 应用程序运行。另一个 Subscriber 应用程序将开始运行,您会看到如下所示的输出:
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
Send Message
在 Publisher 控制台窗口中,输入 Hi 并按回车键以发送消息。
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
Enter message:
Hi
Receive Message
在 Subscriber 控制台窗口中验证,消息在每个窗口中都能收到。
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
Received = Hi
发送退出作为消息以终止所有发布者和订阅者控制台窗口会话。
Verification
现在,在您的浏览器中打开 http://127.0.0.1:8161/admin/ 。它会要求提供凭据。使用 admin/admin 作为用户名/密码,它将加载 ActiveMQ 管理控制台,您可以在其中查看主题以检查状态。它将显示入列和传递的多个消息。