消息中间件ActiveMQ的简单入门介绍与使用

什么是ActiveMQ

ActiveMQ是一款开源的消息中间件，它采用Java编写，完全支持JMS规范，是Apache软件基金会的顶级项目之一。ActiveMQ可以通过网络将不同应用程序之间异构的数据进行传输，是一种比较通用的解决方案。

ActiveMQ的基本概念

在了解如何使用ActiveMQ之前，有几个基本的概念需要了解：

Broker：消息中间件的核心部分，负责管理消息的传递和存储，以及与客户端的通信。
Producer：消息的生产者，向消息中间件发送消息。
Consumer：消息的消费者，从消息中间件接收消息。
Queue：点对点模式下的消息传递方式。
Topic：发布订阅模式下的消息传递方式。

ActiveMQ的安装

ActiveMQ是基于Java开发的，所以安装前需要准备好Java运行环境。具体安装步骤如下：

下载ActiveMQ

可以到ActiveMQ的官方网站https://activemq.apache.org/上下载最新的ActiveMQ安装包，也可以使用命令行下载：

wget https://archive.apache.org/dist/activemq/5.15.8/apache-activemq-5.15.8-bin.tar.gz

解压ActiveMQ

将下载好的安装包解压到指定目录下：

tar -zxvf apache-activemq-5.15.8-bin.tar.gz

启动ActiveMQ

进入apache-activemq-5.15.8/bin目录，执行以下命令启动ActiveMQ：

./activemq start

ActiveMQ的使用

队列模式

在队列模式中，消息生产者向队列发送消息，然后消息消费者从队列中消费消息。

创建队列

在ActiveMQ中，创建队列需要在管理页面操作，具体步骤如下：

访问http://localhost:8161/admin/queues.jsp，在页面下方点击“Create a new Queue”按钮。
在弹出的对话框中填写队列的名称，例如“testQueue”，然后点击“Create”按钮。

发送消息

可以使用Java代码向队列发送消息，示例代码如下：

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class Producer {

    public static void main(String[] args) {
        ConnectionFactory connectionFactory =
                new ActiveMQConnectionFactory("tcp://localhost:61616");

        try (Connection connection = connectionFactory.createConnection()) {
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            Destination destination = session.createQueue("testQueue");

            MessageProducer producer = session.createProducer(destination);

            TextMessage textMessage = session.createTextMessage("Hello Queue!");

            producer.send(textMessage);

            System.out.println("消息发送成功：" + textMessage.getText());
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

}

接收消息

同样可以使用Java代码从队列接收消息，示例代码如下：

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class Consumer {

    public static void main(String[] args) {
        ConnectionFactory connectionFactory =
                new ActiveMQConnectionFactory("tcp://localhost:61616");

        try (Connection connection = connectionFactory.createConnection()) {
            connection.start();

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            Destination destination = session.createQueue("testQueue");

            MessageConsumer consumer = session.createConsumer(destination);

            Message message = consumer.receive();

            TextMessage textMessage = (TextMessage) message;

            System.out.println("接收到的消息：" + textMessage.getText());
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

}

订阅模式

在订阅模式中，消息生产者向主题发送消息，主题会将消息发送给所有订阅该主题的消息消费者。

创建主题

创建主题跟创建队列类似，只需要在管理页面中点击“Create a new Topic”按钮即可。

发送消息

发送消息的Java代码跟队列模式中的代码基本相同，代码如下：

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class Publisher {

    public static void main(String[] args) {
        ConnectionFactory connectionFactory =
                new ActiveMQConnectionFactory("tcp://localhost:61616");

        try (Connection connection = connectionFactory.createConnection()) {
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            Destination destination = session.createTopic("testTopic");

            MessageProducer producer = session.createProducer(destination);

            TextMessage textMessage = session.createTextMessage("Hello Topic!");

            producer.send(textMessage);

            System.out.println("消息发送成功：" + textMessage.getText());
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

}

接收消息

接收消息的代码跟队列模式的代码类似，代码如下：

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class Subscriber {

    public static void main(String[] args) {
        ConnectionFactory connectionFactory =
                new ActiveMQConnectionFactory("tcp://localhost:61616");

        try (Connection connection = connectionFactory.createConnection()) {
            connection.start();

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            Destination destination = session.createTopic("testTopic");

            MessageConsumer consumer = session.createConsumer(destination);

            Message message = consumer.receive();

            TextMessage textMessage = (TextMessage) message;

            System.out.println("接收到的消息：" + textMessage.getText());
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

}