ActiveMQ基础学习简单记录-Toy模板网

这篇具有很好参考价值的文章主要介绍了ActiveMQ基础学习简单记录。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

本文为Activemq简单入门文章

JMS是什么

JMS即Java Message Service，是JavaEE的消息服务接口。JMS主要有两个版本：1.1和2.0。2.0和1.1相比，主要是简化了收发消息的代码。

JMS为Java程序提供了一种通用方法, 用于创建、发送、接收和读取企业消息系统中的消息。

JMS是一组接口定义，如果我们要使用JMS，还需要选择一个具体的JMS产品。常用的JMS服务器有开源的ActiveMQ，商业服务器如WebLogic、WebSphere等也内置了JMS支持。

ActiveMQ基础学习简单记录
使用消息服务，而不是直接调用对方的API，它的好处是：

双方各自无需知晓对方的存在，消息可以异步处理，因为消息服务器会在Consumer离线的时候自动缓存消息；
如果Producer发送的消息频率高于Consumer的处理能力，消息可以积压在消息服务器，不至于压垮Consumer；
通过一个消息服务器，可以连接多个Producer和多个Consumer。

JMS消息模型

JMS提供了两种消息模型:

PTP(点对点消息模型)
Pub/Sub(发布订阅消息模式)
Queue是一种一对一的通道
- 如果Consumer离线无法处理消息时，Queue会把消息存起来，等Consumer再次连接的时候发给它。
- 设定了持久化机制的Queue不会丢失消息。
- 如果有多个Consumer接入同一个Queue，那么它们等效于以集群方式处理消息
  - 例如，发送方发送的消息是A，B，C，D，E，F，两个Consumer可能分别收到A，C，E和B，D，F，即每个消息只会交给其中一个Consumer处理。
Topic则是一种一对多通道。
- 一个Producer发出的消息，会被多个Consumer同时收到，即每个Consumer都会收到一份完整的消息流。
- 如果某个Consumer暂时离线，过一段时间后又上线了，那么在它离线期间产生的消息还能不能收到呢？
  - 这取决于消息服务器对Topic类型消息的持久化机制。
    - 如果消息服务器不存储Topic消息，那么离线的Consumer会丢失部分离线时期的消息，如果消息服务器存储了Topic消息，那么离线的Consumer可以收到自上次离线时刻开始后产生的所有消息。
    - JMS规范通过Consumer指定一个持久化订阅可以在上线后收取所有离线期间的消息，如果指定的是非持久化订阅，那么离线期间的消息会全部丢失。

注意:

如果一个Topic的消息全部都持久化了，并且只有一个Consumer，那么它和Queue其实是一样的。实际上，很多消息服务器内部都只有Topic类型的消息架构，Queue可以通过Topic“模拟”出来。

无论是Queue还是Topic，对Producer没有什么要求。多个Producer也可以写入同一个Queue或者Topic，此时消息服务器内部会自动排序确保消息总是有序的。

上面是消息服务的基本模型。当某个到达消息服务器时，Producer和Consumer通常是通过TCP连接消息服务器，在编写JMS程序时，又会遇到ConnectionFactory、Connection、Session等概念，其实这和JDBC连接是类似的：

ConnectionFactory：代表一个到消息服务器的连接池，类似JDBC的DataSource；
Connection：代表一个到消息服务器的连接，类似JDBC的Connection；
Session：代表一个经过认证后的连接会话；
Message：代表一个消息对象。

在JMS 1.1中，发送消息的代码示例如下：

try {
    Connection connection = null;
    try {
        // 创建连接:
        connection = connectionFactory.createConnection();
        // 创建会话:
        Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
        // 创建一个Producer并关联到某个Queue:
        MessageProducer messageProducer = session.createProducer(queue);
        // 创建一个文本消息:
        TextMessage textMessage = session.createTextMessage(text);
        // 发送消息:
        messageProducer.send(textMessage);
    } finally {
        // 关闭连接:
        if (connection != null) {
            connection.close();
        }
    }
} catch (JMSException ex) {
    // 处理JMS异常
}

JMS 2.0改进了一些API接口，发送消息变得更简单：

try (JMSContext context = connectionFactory.createContext()) {
    context.createProducer().send(queue, text);
}

注意:

JMSContext实现了AutoCloseable接口，可以使用try(resource)语法，代码更简单。

JMS Message Type

JMS的消息类型支持以下几种：

消息类型	适用场景	通信机制	消息传递模型
TextMessage	传递文本消息 String XML JSON	同步或异步	点对点或发布/订阅
MapMessage	传递带有多个属性的消息	同步或异步	点对点或发布/订阅
BytesMessage	传递二进制数据	同步或异步	点对点或发布/订阅
StreamMessage	传递流式数据，适用于需要分块读取数据的场景	同步或异步	点对点或发布/订阅
ObjectMessage	传递Java对象	同步或异步	点对点或发布/订阅
Message	用于接收任意类型的消息，适用于需要动态解析消息类型的场景	同步或异步	点对点或发布/订阅

JMS支持的消息头字段(Message header fields)有:

ActiveMQ基础学习简单记录

Activemq

Activemq简介:

ActiveMQ是一个开源的消息中间件，实现了JMS规范。它提供了很多高级特性，比如消息分组、消息持久化、事务、异步发送等。ActiveMQ支持多种协议和编程语言，可以很方便地与Java、.NET、C++、Ruby等语言进行集成。
在ActiveMQ中，消息由生产者发送到队列或主题，消费者从队列或主题中接收消息。ActiveMQ还提供了许多扩展功能，如消息分组、延迟发送、异步发送等。同时，ActiveMQ也支持多种协议，如STOMP、AMQP、OpenWire等，可以很方便地与其他系统进行集成。
除了基本的JMS功能外，ActiveMQ还提供了许多高级功能，如消息选择器、消息传递规则、消息转发等。此外，ActiveMQ还支持多种消息类型，包括文本、对象、字节流等。
ActiveMQ可以作为单个Java应用程序的嵌入式消息中间件，也可以作为分布式系统的消息中间件。它支持多种部署模式，如集群模式、Master-Slave模式等。
ActiveMQ的插件机制允许通过插件扩展其功能，例如实现消息过滤、路由和安全认证。插件是以Java类的形式存在的，可以通过配置文件或编程的方式来加载它们。ActiveMQ提供了很多可用的插件，比如JAAS身份认证插件、STOMP支持插件、AMQP支持插件等。此外，ActiveMQ还允许自定义插件，以满足特定的需求。

插件:

Camel 插件：将 ActiveMQ 与 Apache Camel 集成，以支持各种数据转换和消息路由模式。

LevelDB 存储插件：使用 LevelDB 作为消息存储的替代方案。

MQTT 插件：支持使用 MQTT 协议进行消息传递。

STOMP 插件：支持使用 STOMP 协议进行消息传递。

Virtual Topics 插件：提供虚拟主题，以简化发布订阅模式的实现。

WebSocket 插件：支持使用 WebSocket 协议进行消息传递。

JMX 监控插件：提供 JMX 监控功能，以监视 ActiveMQ 运行时的性能和状态信息。

安装

ActiveMQ Classic或者ActiveMQ Artemis的关系:

ActiveMQ Classic原来就叫ActiveMQ，是Apache开发的基于JMS 1.1的消息服务器，目前稳定版本号是5.x
ActiveMQ Artemis是由RedHat捐赠的HornetQ服务器代码的基础上开发的，目前稳定版本号是2.x。
- 和ActiveMQ Classic相比，Artemis版的代码与Classic完全不同，并且，它支持JMS 2.0，使用基于Netty的异步IO，大大提升了性能。
- 此外，Artemis不仅提供了JMS接口，它还提供了AMQP接口，STOMP接口和物联网使用的MQTT接口。选择Artemis，相当于一鱼四吃。

所以，我们这里直接选择ActiveMQ Artemis。从官网下载最新的2.x版本，解压后设置环境变量ARTEMIS_HOME，指向Artemis根目录，例如C:\Apps\artemis，然后，把ARTEMIS_HOME/bin加入PATH环境变量：

Windows下添加%ARTEMIS_HOME%\bin到Path路径；
Mac和Linux下添加$ARTEMIS_HOME/bin到PATH路径。

Artemis有个很好的设计，就是它把程序和数据完全分离了。我们解压后的ARTEMIS_HOME目录是程序目录，要启动一个Artemis服务，还需要创建一个数据目录。我们把数据目录直接设定在项目spring-integration-jms的jms-data目录下。执行命令artemis create jms-data：

ActiveMQ基础学习简单记录

在创建过程中，会要求输入连接用户和口令，这里我们设定admin和password，以及是否允许匿名访问（这里选择N）。

此数据目录jms-data不仅包含消息数据、日志，还自动创建了两个启动服务的命令bin/artemis和bin/artemis-service，前者在前台启动运行，按Ctrl+C结束，后者会一直在后台运行。

我们把目录切换到jms-data/bin，直接运行artemis run即可启动Artemis服务：

ActiveMQ基础学习简单记录
启动成功后，Artemis提示可以通过URL: http://localhost:8161/console访问管理后台。注意不要关闭命令行窗口。

如果Artemis启动时显示警告：AMQ222212: Disk Full! … Clients will report blocked.这是因为磁盘空间不够，可以在etc/broker.xml配置中找到<max-disk-usage>并改为99。

ActiveMQ基础学习简单记录

概念强化

上面简单介绍了JMS的基本概念，本节结合Activemq来具体说明JMS的使用。

JMS中的核心概念:

JMS Client：用来发送和接收消息的Java程序
Non-JMS client：不使用 JMS API 的客户端应用程序。这意味着这些应用程序不使用 JMS 提供的接口和协议与消息中间件进行交互。

注意:

Non-JMS client 可能使用其他协议或方式与消息中间件进行通信，例如: 直接使用底层的消息中间件提供的原生 API、使用自定义的消息格式或协议进行通信等。

这就好像你不使用java提供的JBDC统一驱动接口进行调用，而直接调用各个第三方厂商提供的驱动实现类一样

需要注意的是，使用 Non-JMS client 进行消息通信可能会导致与特定消息中间件的耦合性增加。因为它们直接依赖于消息中间件提供的接口和协议，所以在切换或迁移到其他消息中间件时可能需要进行修改和适配

Non-JMS client 的存在也提供了一种灵活性和自由度，可以根据特定需求选择更适合的通信方式和协议。它们可能适用于特定的应用场景或需要与非 JMS 兼容的系统进行集成的情况。

Messages：JMS Client之间进行通讯的消息
JMS Provider (JMS提供者)：实现了JMS规范的消息系统，提供了消息服务和管理功能

类比: mysql-connector-java

Administered Objects(被管理的对象)：是预先配置的JMS对象，通常由管理员或开发者创建 --> ConnectionFactory和Destination(可以理解为队列Queue)

注意:

在JMS（Java Message Service）中，Administered Objects（管理对象）是由JMS提供者（如消息中间件）管理和提供的一些资源，包括队列（Queue）、主题（Topic）、连接工厂（ConnectionFactory）等。这些对象提供了与消息传递相关的基础设施，并允许应用程序与消息中间件进行交互。

通过使用 Administered Objects，应用程序可以更方便地与消息中间件进行交互，而无需了解底层的通信协议和细节。应用程序可以通过配置或通过编程的方式访问和使用这些对象，以满足不同的消息传递需求

引入相关依赖:

        <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>artemis-jms-client</artifactId>
            <version>2.11.0</version>
        </dependency>
         <dependency>
            <groupId>javax.json</groupId>
            <artifactId>javax.json-api</artifactId>
            <version>1.1.4</version>
        </dependency>
        <dependency>
            <groupId>org.glassfish</groupId>
            <artifactId>javax.json</artifactId>
            <version>1.1.4</version>
        </dependency>

低版本的Artemis的Client依赖包中,没有下面这个问题，无需引入。
高版本的Artemis的Client依赖包中，虽然依赖了jakarta.jms:jakarta.jms-api，但是由于artemis-jakarta-client包依赖的jakarta.jms-api版本比较老，内部包名还是旧版本的javax.jms开头:

因此需要手动引入高版本jakarta.jms-api，以maven依赖的就近查找原则，覆盖旧版本:

上面已经介绍过了JMS中支持的消息类型,下面我们来看一下如何创建并发送不同类型的消息:

发布一条不含有任何消息内容的异常文本消息 – 不含有有效负载的简单通知，仅包含JSM消息头和消息属性，用于事件通知

    public Message createMessage(Session session) throws JMSException {
        Message m = session.createMessage();
        m.setStringProperty("exception", "java.lang.NosuchMethodException");
        return m;
    }

发布一条仅包含文本类型的消息以及XML和JSON

    public Message createMessage(Session session) throws JMSException {
        TextMessage tm = session.createTextMessage();
        tm.setText("{ \"name\": \"John\", \"age\": 30}");
        return tm;
    }

发布一组key/value作为消息内容

    public Message createMessage(Session session) throws JMSException {
        MapMessage mm = session.createMapMessage();
        mm.setInt("id", 2017);
        mm.setString("name", "zhangsan");
        mm.setString("password", "123456");
        return mm;
    }

发布一组二进制数组作为消息内容 — 可以传输任意二进制类型的数据,如: 图像和音频

        MessageProducer producer = session.createProducer(queue);
        BytesMessage message = session.createBytesMessage();
        // 设置消息内容
        byte[] payload = "Hello, JMS!".getBytes();
        message.writeBytes(payload);
        // 发送消息
        producer.send(message);

发布可序列化的java对象作为消息内容

    public Message createMessage(Session session) throws JMSException {
        ObjectMessage om = session.createObjectMessage();
        User user = new User();
        user.setID(2020001);
        user.setName("zhangsan");
        om.setObject(user);
        return om;
    }

传递流式数据，用于处理连续数据流的消息类型 --> 如: 用于传输和处理来自传感器等源的实时数据

生产者:

import jakarta.jms.*;
import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;

public class Producer {
    public static void main(String[] args) {
        try {
            // 创建连接工厂和连接
 ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616","root","123456");
            Connection connection = connectionFactory.createConnection();
            connection.start();

            // 创建会话和队列
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Destination destination = session.createQueue("test");

            // 创建生产者并发送多个消息
            MessageProducer producer = session.createProducer(destination);

            for (int i = 0; i < 5; i++) {
                // 创建消息并设置消息内容
                StreamMessage message = session.createStreamMessage();
                message.writeInt(i);
                message.writeString("Message " + i);

                // 发送消息
                producer.send(message);
                System.out.println("Sent message: " + message);
            }

            // 关闭连接和会话
            producer.close();
            session.close();
            connection.close();
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

消费者:

import jakarta.jms.*;
import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;

public class Consumer {
    public static void main(String[] args) {
        try {
            // 创建连接工厂和连接
 ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616","root","123456");
            Connection connection = connectionFactory.createConnection();
            connection.start();

            // 创建会话和队列
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Destination destination = session.createQueue("test");

            // 创建消费者并接收消息
            MessageConsumer consumer = session.createConsumer(destination);

            while (true) {
                Message message = consumer.receive();
                if (message instanceof StreamMessage) {
                    StreamMessage streamMessage = (StreamMessage) message;
                    int intValue = streamMessage.readInt();
                    String stringValue = streamMessage.readString();
                    System.out.println("Received message: " + intValue + ", " + stringValue);
                }else {
                    break;
                }
            }

            // 关闭连接和会话
            consumer.close();
            session.close();
            connection.close();
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

JMS的跨平台性

ActiveMQ基础学习简单记录
JMS的跨平台实现与JDBC类似，核心思路是如何定位到第三方厂商提供的服务实现类。

就像JDBC中，我们需要通过JDK SPI机制获取Drive接口实现类一样，JMS同样需要采用一种服务发现机制，获取第三方提供的:

ConnectionFactory：客户端用来创建同JMS提供者之间的连接的对象
Destination ：客户端用来指明消息被发送的目的地以及客户端接收消息的来源 --> Queue

被管理的对象一般被管理员放在JNDI名字空间中，通常在JMS客户端应用的文档中说明它所需要的JMS被管理对象，以及应以何种JNDI名字来提供这些JMS被管理对象。

JMS通用接口

ActiveMQ基础学习简单记录
JMS 1.1中各个接口之间的关系如下图所示:

ActiveMQ基础学习简单记录
JMS 2.0中改进了一些API接口，此时各个接口之间的关系如下所示:

ActiveMQ基础学习简单记录
在JMS（Java消息服务）的简化API中，单个JMSContext对象包含了经典API中由两个独立对象提供的行为，即Connection和Session。尽管该规范中提到了JMSContext具有底层的“连接”和“会话”，但简化API并不使用Connection和Session接口。

在简化API中，JMSContext作为一个统一的入口点，用于创建JMS对象、发送和接收消息以及管理事务。它以更简洁和方便的方式封装了连接和会话的行为。

通过使用JMSContext，可以执行诸如创建生产者或消费者、发送和接收消息、提交或回滚事务等操作。JMSContext在后台管理底层的连接和会话，提供了一个简化和更直观的编程模型。

简化API的目标是使JMS的使用更加简单和直观，减少开发人员直接使用底层连接和会话对象的需求。相反，他们可以依赖于JMSContext来处理这些细节，集中精力进行核心消息操作。

总之，JMSContext是JMS简化API中的主要对象，它整合了连接和会话的功能，并提供了更简洁、易用的编程模型，使JMS的使用更加便捷。

下面列举一个demo示例:

生产者:

package org.example;

import javax.jms.*;

public class JMSProducer {

    public static void main(String[] args) {
        // 设置 ActiveMQ 的连接信息
        String brokerUrl = "tcp://localhost:61616";
        String username = "root";
        String password = "123456";
        String queueName = "jms/queue/mail";

        // 创建连接工厂
        ConnectionFactory connectionFactory = new org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory(brokerUrl);

        try (JMSContext context = connectionFactory.createContext(username, password)) {
            // 创建目标队列
            Destination destination = context.createQueue(queueName);

            // 创建消息生产者
            javax.jms.JMSProducer producer = context.createProducer();

            // 创建文本消息
            TextMessage message = context.createTextMessage();
            message.setText("Hello, JMS!");

            // 发送消息
            producer.send(destination, message);

            System.out.println("Message sent successfully.");
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

消费者:

package org.example;

import javax.jms.*;

public class JMSConsumer {

    public static void main(String[] args) {
        // 设置 ActiveMQ 的连接信息
        String brokerUrl = "tcp://localhost:61616";
        String username = "root";
        String password = "123456";
        String queueName = "jms/queue/mail";

        // 创建连接工厂
        ConnectionFactory connectionFactory = new org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory(brokerUrl);

        try (JMSContext context = connectionFactory.createContext(username, password)) {
            // 创建目标队列
            Destination destination = context.createQueue(queueName);

            // 创建消息消费者
            javax.jms.JMSConsumer consumer = context.createConsumer(destination);

            // 接收消息
            Message message = consumer.receive();

            if (message instanceof TextMessage) {
                TextMessage textMessage = (TextMessage) message;
                System.out.println("Received message: " + textMessage.getText());
            } else {
                System.out.println("Received message of unsupported type.");
            }
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

JMS希望达到的目标是什么

目标1：JMS提供一种标准的、平台无关的方法，使分布式应用程序之间可以可靠地交换消息。

JMS允许应用程序发送和接收消息，这些消息可以是文本消息、字节消息或对象消息，可以在同步或异步模式下进行发送和接收。
JMS 还提供了一些高级特性，如消息持久化、应答机制、事务支持和消息过滤，以满足不同应用程序的需求。

目标2：实现松耦合的消息传递机制，使得分布式应用程序能够通过消息的方式进行通信，而不需要知道对方的具体实现。

这种松耦合的通信机制使得分布式应用程序能够更加灵活、可扩展、可维护，同时也能提高应用程序的可靠性和可用性。

Activemq发送消息的三种模式

至少一次

At least once 模式确保消息至少被传递一次，但是可能会被传递多次。

在这种模式下，消息消费者在接收到消息后必须向ActiveMQ发送一个确认消息，告知ActiveMQ消息已经被正确接收并处理。
如果消费者没有发送确认消息，则ActiveMQ会认为消息未被正确处理，尝试重新传递消息，直到收到确认消息为止。
如果消费者发送的ack丢失，会导致Broker重新发送消息。
该模式下需要注意消息的幂等性

以下是使用 ActiveMQ 实现 “at least once” 语义的消息消费者代码示例:

package org.example;

import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;

import javax.jms.*;

public class JMSConsumer {

    public static void main(String[] args) {
        String brokerUrl = "tcp://localhost:61616";
        String username = "your-username";
        String password = "your-password";
        String queueName = "jms/queue/mail";

        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(brokerUrl);

        try (Connection connection = connectionFactory.createConnection(username, password)) {
            connection.start();
            //CLIENT_ACKNOWLEDGE：表示消费者receive消息后必须手动的调用acknowledge()方法进行签收
            Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
            Destination destination = session.createQueue(queueName);
            MessageConsumer consumer = session.createConsumer(destination);

            consumer.setMessageListener(new MessageListener() {
                public void onMessage(Message message) {
                    try {
                        if (message instanceof TextMessage) {
                            TextMessage textMessage = (TextMessage) message;
                            String text = textMessage.getText();
                            System.out.println("Received message: " + text);
                            // 执行具体的业务逻辑

                            // 手动确认消息已被处理
                            message.acknowledge();
                        }
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });

            // 暂停主线程，等待消息到达和处理
            Thread.sleep(30000);

            session.close();
            connection.close();
        } catch (JMSException | InterruptedException e) {
            e.printStackTrace();
        }
    }
}

至多一次

At most once模式确保消息最多被传递一次。

在这种模式下，消息消费者在接收到消息后不需要发送确认消息，ActiveMQ会假设消息已经被正确处理，不会尝试重新传递消息

以下是一个使用 ActiveMQ 的消息消费者代码示例，展示了 “at most once” 的行为：

package org.example;

import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;

import javax.jms.*;

public class JMSConsumer {

    public static void main(String[] args) {
        String brokerUrl = "tcp://localhost:61616";
        String username = "your-username";
        String password = "your-password";
        String queueName = "jms/queue/mail";

        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(brokerUrl);

        try (Connection connection = connectionFactory.createConnection(username, password)) {
            connection.start();
            //AUTO_ACKNOWLEDGE：表示在消费者receive消息的时候自动的签收    
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Destination destination = session.createQueue(queueName);
            MessageConsumer consumer = session.createConsumer(destination);

            consumer.setMessageListener(new MessageListener() {
                public void onMessage(Message message) {
                    try {
                        if (message instanceof TextMessage) {
                            TextMessage textMessage = (TextMessage) message;
                            String text = textMessage.getText();
                            System.out.println("Received message: " + text);
                            // 执行具体的业务逻辑
                        }
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });

            // 暂停主线程，等待消息到达和处理
            Thread.sleep(30000);

            session.close();
            connection.close();
        } catch (JMSException | InterruptedException e) {
            e.printStackTrace();
        }
    }
}

精确一次

Exactly once模式确保消息只被传递一次。

要实现消息的精确一次发送（Exactly Once），可以使用 ActiveMQ 提供的事务机制来确保消息的可靠性传递。

以下是使用 ActiveMQ 实现精确一次发送语义的消息生产者和消费者代码示例：

生产者:

public class JMSProducer {

    public static void main(String[] args) {
        String brokerUrl = "tcp://localhost:61616";
        String username = "your-username";
        String password = "your-password";
        String queueName = "jms/queue/mail";

        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(brokerUrl);

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

            Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
            Destination destination = session.createQueue(queueName);
            MessageProducer producer = session.createProducer(destination);

            // 创建并发送消息
            TextMessage message = session.createTextMessage("Hello, ActiveMQ!");
            producer.send(message);

            // 提交事务
            session.commit();

            session.close();
            connection.close();
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

消费者:

public class JMSConsumer {

    public static void main(String[] args) {
        String brokerUrl = "tcp://localhost:61616";
        String username = "your-username";
        String password = "your-password";
        String queueName = "jms/queue/mail";

        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(brokerUrl);

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

            Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
            Destination destination = session.createQueue(queueName);
            MessageConsumer consumer = session.createConsumer(destination);

            consumer.setMessageListener(new MessageListener() {
                public void onMessage(Message message) {
                    try {
                        if (message instanceof TextMessage) {
                            TextMessage textMessage = (TextMessage) message;
                            String text = textMessage.getText();
                            System.out.println("Received message: " + text);

                            // 执行具体的业务逻辑

                            // 手动提交事务
                            session.commit();
                        }
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });

            // 暂停主线程，等待消息到达和处理
            Thread.sleep(30000);

            session.close();
            connection.close();
        } catch (JMSException | InterruptedException e) {
            e.printStackTrace();
        }
    }
}

在这个代码示例中，我们创建了一个消息生产者和一个消息消费者。消息生产者使用 Session.SESSION_TRANSACTED 模式创建了会话，并在发送消息后提交了事务。消息消费者也使用 Session.SESSION_TRANSACTED 模式创建了会话，并在处理消息后手动提交了事务。

通过使用事务机制，可以确保消息在发送和接收过程中的可靠性。如果消息发送或处理过程中发生异常，事务会回滚，消息不会被确认，从而实现了消息的精确一次发送语义（Exactly Once）。

ActiveMQ的事务机制的底层原理涉及到消息的持久化和日志的记录。

当使用事务提交时，ActiveMQ会将事务中的消息写入持久化存储，通常是磁盘上的数据库或文件系统。这样可以确保在发生故障或断电等情况下，消息的持久性得到保证，不会丢失。

同时，ActiveMQ会将事务的操作记录在事务日志中。事务日志记录了所有发送、接收和确认消息的操作，以及事务的状态信息。这样可以在系统恢复时，根据事务日志的内容来恢复之前未完成的事务，并确保事务的一致性。

事务提交的过程可以简述为以下几个步骤：

在事务提交时，将事务中的消息写入持久化存储。

将事务的操作记录写入事务日志。

标记事务为已提交。

释放事务相关资源。

在事务回滚的情况下，会根据事务日志中的记录进行回滚操作，包括将持久化存储中的消息删除或标记为未发送状态，并将事务标记为已回滚。

通过持久化存储和事务日志的机制，ActiveMQ能够提供消息的可靠性传递和事务的原子性，确保消息在发送和接收过程中的可靠性和一致性。这样可以保证消息在分布式系统中的可靠性传递，并提供消息的精确一次发送语义（Exactly
Once）。

可重复确认模式

DUPS_OK_ACKNOWLEDGE（重复确认模式）适用于消息重复消费不会造成严重问题的场景，

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

需要注意的是，由于DUPS_OK_ACKNOWLEDGE模式下允许消息的重复消费，因此不能将该模式应用于对数据准确性要求极高的场景，例如金融交易等。

小结

此部分内容转载此文: Activemq中的四种消息确认模式

AUTO_ACKNOWLEDGE : 自动确认,这就意味着消息的确认时机将由consumer择机确认."择机确认"似乎充满了不确定性,这也意味着,开发者必须明确知道"择机确认"的具体时机,否则将有可能导致消息的丢失,或者消息的重复接受.那么在ActiveMQ中,AUTO_ACKNOWLEDGE是如何运作的呢?

对于consumer而言，optimizeAcknowledge属性只会在AUTO_ACK模式下有效。
其中DUPS_ACKNOWLEGE也是一种潜在的AUTO_ACK,只是确认消息的条数和时间上有所不同。
在“同步”(receive)方法返回message之前,会检测optimizeACK选项是否开启，如果没有开启，此单条消息将立即确认，所以在这种情况下，message返回之后，如果开发者在处理message过程中出现异常，会导致此消息也不会redelivery,即"潜在的消息丢失"；如果开启了optimizeACK，则会在unAck数量达到prefetch * 0.65时确认，当然我们可以指定prefetchSize = 1来实现逐条消息确认。
在"异步"(messageListener)方式中,将会首先调用listener.onMessage(message),此后再ACK,如果onMessage方法异常,将导致client端补充发送一个ACK_TYPE为REDELIVERED_ACK_TYPE确认指令；如果onMessage方法正常,消息将会正常确认(STANDARD_ACK_TYPE)。此外需要注意，消息的重发次数是有限制的，每条消息中都会包含“redeliveryCounter”计数器，用来表示此消息已经被重发的次数，如果重发次数达到阀值，将会导致发送一个ACK_TYPE为POSION_ACK_TYPE确认指令,这就导致broker端认为此消息无法消费,此消息将会被删除或者迁移到"dead letter"通道中。

因此当我们使用messageListener方式消费消息时，通常建议在onMessage方法中使用try-catch,这样可以在处理消息出错时记录一些信息，而不是让consumer不断去重发消息；如果你没有使用try-catch,就有可能会因为异常而导致消息重复接收的问题,需要注意你的onMessage方法中逻辑是否能够兼容对重复消息的判断。

CLIENT_ACKNOWLEDGE : 客户端手动确认，这就意味着AcitveMQ将不会“自作主张”的为你ACK任何消息，开发者需要自己择机确认。

在此模式下，开发者需要需要关注几个方法：

1. message.acknowledge()，
2. ActiveMQMessageConsumer.acknowledege()，
3. ActiveMQSession.acknowledge()；

其中1和3是等效的，将当前session中所有consumer中尚未ACK的消息都一起确认，2只会对当前consumer中那些尚未确认的消息进行确认。开发者可以在合适的时机必须调用一次上述方法。
我们通常会在基于Group(消息分组)情况下会使用CLIENT_ACKNOWLEDGE，我们将在一个group的消息序列接受完毕之后确认消息(组)；不过当你认为消息很重要，只有当消息被正确处理之后才能确认时，也可以使用此ACK_MODE。
如果开发者忘记调用acknowledge方法，将会导致当consumer重启后，会接受到重复消息，因为对于broker而言，那些尚未真正ACK的消息被视为“未消费”。
开发者可以在当前消息处理成功之后，立即调用message.acknowledge()方法来"逐个"确认消息，这样可以尽可能的减少因网络故障而导致消息重发的个数；当然也可以处理多条消息之后，间歇性的调用acknowledge方法来一次确认多条消息，减少ack的次数来提升consumer的效率，不过这仍然是一个利弊权衡的问题。
除了message.acknowledge()方法之外，ActiveMQMessageConumser.acknowledge()和ActiveMQSession.acknowledge()也可以确认消息，只不过前者只会确认当前consumer中的消息。其中sesson.acknowledge()和message.acknowledge()是等效的。
无论是“同步”/“异步”，ActiveMQ都不会发送STANDARD_ACK_TYPE，直到message.acknowledge()调用。如果在client端未确认的消息个数达到prefetchSize * 0.5时，会补充发送一个ACK_TYPE为DELIVERED_ACK_TYPE的确认指令，这会触发broker端可以继续push消息到client端。(参看PrefetchSubscription.acknwoledge方法)
在broker端，针对每个Consumer，都会保存一个因为"DELIVERED_ACK_TYPE"而“拖延”的消息个数，这个参数为prefetchExtension，事实上这个值不会大于prefetchSize * 0.5,因为Consumer端会严格控制DELIVERED_ACK_TYPE指令发送的时机(参见ActiveMQMessageConsumer.ackLater方法)，broker端通过“prefetchExtension”与prefetchSize互相配合，来决定即将push给client端的消息个数，count = prefetchExtension + prefetchSize - dispatched.size()，其中dispatched表示已经发送给client端但是还没有“STANDARD_ACK_TYPE”的消息总量；
由此可见，在CLIENT_ACK模式下，足够快速的调用acknowledge()方法是决定consumer端消费消息的速率；如果client端因为某种原因导致acknowledge方法未被执行，将导致大量消息不能被确认，broker端将不会push消息，事实上client端将处于“假死”状态，而无法继续消费消息。我们要求client端在消费1.5*prefetchSize个消息之前，必须acknowledge()一次；通常我们总是每消费一个消息调用一次，这是一种良好的设计。
此外需要额外的补充一下：所有ACK指令都是依次发送给broker端，在CLIET_ACK模式下，消息在交付给listener之前，都会首先创建一个DELIVERED_ACK_TYPE的ACK指令，直到client端未确认的消息达到"prefetchSize * 0.5"时才会发送此ACK指令，如果在此之前，开发者调用了acknowledge()方法，会导致消息直接被确认(STANDARD_ACK_TYPE)。broker端通常会认为“DELIVERED_ACK_TYPE”确认指令是一种“slow consumer”信号，如果consumer不能及时的对消息进行acknowledge而导致broker端阻塞，那么此consumer将会被标记为“slow”，此后queue中的消息将会转发给其他Consumer。

DUPS_OK_ACKNOWLEDGE : "消息可重复"确认，意思是此模式下，可能会出现重复消息，并不是一条消息需要发送多次ACK才行。它是一种潜在的"AUTO_ACK"确认机制，为批量确认而生，而且具有“延迟”确认的特点。对于开发者而言，这种模式下的代码结构和AUTO_ACKNOWLEDGE一样，不需要像CLIENT_ACKNOWLEDGE那样调用acknowledge()方法来确认消息。

在ActiveMQ中，如果所在Destination是Queue通道，我们真的可以认为DUPS_OK_ACK就是“AUTO_ACK + optimizeACK + (prefetch > 0)”这种情况，在确认时机上几乎完全一致；此外在此模式下，如果prefetchSize =1 或者没有开启optimizeACK，也会导致消息逐条确认，从而失去批量确认的特性。
如果Destination为Topic，DUPS_OK_ACKNOWLEDGE才会产生JMS规范中诠释的意义，即无论optimizeACK是否开启，都会在消费的消息个数>=prefetch * 0.5时，批量确认(STANDARD_ACK_TYPE),在此过程中，不会发送DELIVERED_ACK_TYPE的确认指令,这是和AUTO_ACK的最大的区别。
这也意味着，当consumer故障重启后，那些尚未ACK的消息会重新发送过来。

SESSION_TRANSACTED : 当session使用事务时，就是使用此模式。在事务开启之后，和session.commit()之前，所有消费的消息，要么全部正常确认，要么全部redelivery。这种严谨性，通常在基于GROUP(消息分组)或者其他场景下特别适合。在SESSION_TRANSACTED模式下，optimizeACK并不能发挥任何效果,因为在此模式下，optimizeACK会被强制设定为false，不过prefetch仍然可以决定DELIVERED_ACK_TYPE的发送时机。

因为Session非线程安全，那么当前session下所有的consumer都会共享同一个transactionContext；同时建议，一个事务类型的Session中只有一个Consumer，已避免rollback()或者commit()方法被多个consumer调用而造成的消息混乱。
当consumer接受到消息之后，首先检测TransactionContext是否已经开启，如果没有，就会开启并生成新的transactionId，并把信息发送给broker；此后将检测事务中已经消费的消息个数是否 >= prefetch * 0.5,如果大于则补充发送一个“DELIVERED_ACK_TYPE”的确认指令；这时就开始调用onMessage()方法，如果是同步(receive),那么即返回message。上述过程，和其他确认模式没有任何特殊的地方。
当开发者决定事务可以提交时，必须调用session.commit()方法，commit方法将会导致当前session的事务中所有消息立即被确认；事务的确认过程中，首先把本地的deliveredMessage队列中尚未确认的消息全部确认(STANDARD_ACK_TYPE)；此后向broker发送transaction提交指令并等待broker反馈，如果broker端事务操作成功，那么将会把本地deliveredMessage队列清空，新的事务开始；如果broker端事务操作失败(此时broker已经rollback)，那么对于session而言，将执行inner-rollback，这个rollback所做的事情，就是将当前事务中的消息清空并要求broker重发(REDELIVERED_ACK_TYPE),同时commit方法将抛出异常。
当session.commit方法异常时，对于开发者而言通常是调用session.rollback()回滚事务(事实上开发者不调用也没有问题)，当然你可以在事务开始之后的任何时机调用rollback(),rollback意味着当前事务的结束，事务中所有的消息都将被重发。需要注意，无论是inner-rollback还是调用session.rollback()而导致消息重发，都会导致message.redeliveryCounter计数器增加，最终都会受限于brokerUrl中配置的"jms.redeliveryPolicy.maximumRedeliveries",如果rollback的次数过多，而达到重发次数的上限时，消息将会被DLQ(dead letter)。

INDIVIDUAL_ACKNOWLEDGE : 单条消息确认，这种确认模式，我们很少使用，它的确认时机和CLIENT_ACKNOWLEDGE几乎一样，当消息消费成功之后，需要调用message.acknowledege来确认此消息(单条)，而CLIENT_ACKNOWLEDGE模式先message.acknowledge()方法将导致整个session中所有消息被确认(批量确认)。

Activemq支持众多协议

这里以In Jvm协议进行讲解:

ActiveMQ在内部支持以下的In-JVM协议：

VM Transport Protocol（VM传输协议）：这是ActiveMQ默认的In-JVM协议，它允许在同一个Java虚拟机（JVM）内的不同线程之间进行消息的传递。使用VM传输协议，消息可以在应用程序内部的不同组件之间快速传递，而无需通过网络进行通信。
VM Transport Bridge（VM传输桥）：这是ActiveMQ提供的用于连接不同Java虚拟机（JVM）之间的In-JVM通信的协议。它允许在不同的JVM实例之间通过内存进行消息传递，提供了一种简单而高效的方式来实现进程间通信。

这些In-JVM协议适用于在同一个Java虚拟机内的不同线程或不同进程之间进行高性能的消息传递。它们可以避免使用网络通信带来的延迟和开销，提供了更快的消息传递速度和更低的资源消耗。

ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("vm://localhost");
factory.setObjectMessageSerializationDefered(true);

Activemq还支持AMQP协议，MQTT协议等，这里不多列举了。

Activemq支持的定时消息,延迟消息,优先级消息

这里简单介绍一下，具体配置和使用说明，大家参考官方文档或自行查询资料学习。
ActiveMQ基础学习简单记录

延迟消息

MessageProducer producer = session.createProducer(destination);
TextMessage message = session.createTextMessage("test msg");
long time = 60 * 1000;
message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, time);
producer.send(message);

您可以将消息设置为在初始延迟时等待，重复传递 10 次，每次重新传递之间等待 10 秒：

MessageProducer producer = session.createProducer(destination);
TextMessage message = session.createTextMessage("test msg");
long delay = 30 * 1000;
long period = 10 * 1000;
int repeat = 9;
message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, delay);
message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_PERIOD, period);
message.setIntProperty(ScheduledMessage.AMQ_SCHEDULED_REPEAT, repeat);
producer.send(message);

通过CRON表达式设置消息被间隔发送一次

MessageProducer producer = session.createProducer(destination);
TextMessage message = session.createTextMessage("test msg");
message.setStringProperty(ScheduledMessage.AMQ_SCHEDULED_CRON, "0 * * * *");
producer.send(message);

假设您希望一条消息传递 10 次，每条消息之间有一秒钟的延迟 - 并且您希望每小时发生一次 - 您将这样做：

MessageProducer producer = session.createProducer(destination);
TextMessage message = session.createTextMessage("test msg");
message.setStringProperty(ScheduledMessage.AMQ_SCHEDULED_CRON, "0 * * * *");
message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, 1000);
message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_PERIOD, 1000);
message.setIntProperty(ScheduledMessage.AMQ_SCHEDULED_REPEAT, 9);
producer.send(message);

消息选择器

ActiveMQ 是一个流行的开源消息中间件，它支持多种消息选择器的方式。消息选择器允许您从消息队列中选择特定的消息，以便只有满足某些条件的消息会被消费者接收。

在 ActiveMQ 中，消息选择器使用 SQL-92 类似的语法来定义选择条件。您可以在创建消费者时使用消息选择器，通过在消息选择器表达式中指定条件来选择感兴趣的消息。

下面是一些常见的消息选择器示例：

使用消息属性进行选择：

String selector = "color = 'red'";
MessageConsumer consumer = session.createConsumer(destination, selector);

上述代码中，消息选择器指定了一个属性 color 的值为 'red' 的消息将被选择。

使用 JMS 头部属性进行选择：

String selector = "JMSPriority > 5";
MessageConsumer consumer = session.createConsumer(destination, selector);

上述代码中，消息选择器指定了一个 JMS 头部属性 JMSPriority 的值大于 5 的消息将被选择。

使用消息体内容进行选择：

String selector = "text LIKE '%important%'";
MessageConsumer consumer = session.createConsumer(destination, selector);

上述代码中，消息选择器指定了一个消息体内容包含 'important' 关键字的消息将被选择。

请注意，消息选择器只能应用于支持消息选择器功能的消息中间件，而且消费者必须使用带有消息选择器的 createConsumer 方法来创建。另外，使用过多的消息选择器可能会对系统性能产生负面影响，因此在使用时需要权衡选择条件的复杂性和性能需求。

这里只给出一个简单的Demo,关于消息选择器的更多知识，可以参考官方文档，或者自行查找资料学习:

生产者:

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class MessageProducerExample {
    public static void main(String[] args) throws JMSException {
        // 连接到 ActiveMQ 代理
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
        Connection connection = connectionFactory.createConnection();
        connection.start();

        // 创建会话
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        // 创建目标队列
        Destination destination = session.createQueue("myQueue");

        // 创建生产者
        MessageProducer producer = session.createProducer(destination);

        // 创建消息
        TextMessage message1 = session.createTextMessage();
        message1.setText("Message 1");
        message1.setStringProperty("color", "red");

        TextMessage message2 = session.createTextMessage();
        message2.setText("Message 2");
        message2.setStringProperty("color", "blue");

        // 发送消息
        producer.send(message1);
        producer.send(message2);

        // 关闭连接
        connection.close();
    }
}

消费者:

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class MessageConsumerExample {
    public static void main(String[] args) throws JMSException {
        // 连接到 ActiveMQ 代理
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
        Connection connection = connectionFactory.createConnection();
        connection.start();

        // 创建会话
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        // 创建目标队列
        Destination destination = session.createQueue("myQueue");

        // 创建消费者并设置消息选择器
        MessageConsumer consumer = session.createConsumer(destination, "color = 'red'");

        // 接收消息
        Message message = consumer.receive();
        if (message instanceof TextMessage) {
            TextMessage textMessage = (TextMessage) message;
            System.out.println("Received message: " + textMessage.getText());
        }

        // 关闭连接
        connection.close();
    }
}

上述示例中，生产者向名为 “myQueue” 的队列发送了两条消息，每条消息都带有一个名为 “color” 的属性。然后消费者通过设置消息选择器 “color = ‘red’”，只接收具有红色属性的消息。

Spring整合

此部分内容转载: 廖雪峰大佬的JMS入门教程

在上面依赖基础上，引入如下依赖

      <dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-jms</artifactId>
      </dependency>

在AppConfig中，通过@EnableJms让Spring自动扫描JMS相关的Bean，并加载JMS配置文件jms.properties：

@EnableJms
@PropertySource({"classpath:/jms.properties"})
@SpringBootApplication
public class AppConfig {
    public static void main(String[] args) {
        SpringApplication.run(AppConfig.class, args);
    }
}

首先要创建的Bean是ConnectionFactory，即连接消息服务器的连接池：

    @Bean
    ConnectionFactory createJMSConnectionFactory(
            @Value("${jms.uri:tcp://localhost:61616}") String uri,
            @Value("${jms.username:admin}") String username,
            @Value("${jms.password:password}") String password) {
        return new ActiveMQJMSConnectionFactory(uri, username, password);
    }

因为我们使用的消息服务器是ActiveMQ Artemis，所以ConnectionFactory的实现类就是消息服务器提供的ActiveMQJMSConnectionFactory，它需要的参数均由配置文件读取后传入，并设置了默认值。

我们再创建一个JmsTemplate，它是Spring提供的一个工具类，和JdbcTemplate类似，可以简化发送消息的代码：

    @Bean
    JmsTemplate createJmsTemplate(@Autowired ConnectionFactory connectionFactory) {
        return new JmsTemplate(connectionFactory);
    }

下一步要创建的是JmsListenerContainerFactory，

    @Bean("jmsListenerContainerFactory")
    DefaultJmsListenerContainerFactory createJmsListenerContainerFactory(@Autowired ConnectionFactory connectionFactory) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        return factory;
    }

除了必须指定Bean的名称为jmsListenerContainerFactory外，这个Bean的作用是处理和Consumer相关的Bean。我们先跳过它的原理，继续编写MessagingService来发送消息：

@Component
public class MessagingService {
    @Autowired
    private ObjectMapper objectMapper;
    @Autowired
    private JmsTemplate jmsTemplate;

    public void sendMailMessage(MailMessage msg) throws Exception {
        final String text = objectMapper.writeValueAsString(msg);
        jmsTemplate.send("jms/queue/mail", new MessageCreator() {
            public Message createMessage(Session session) throws JMSException {
                return session.createTextMessage(text);
            }
        });
    }

    public static class MailMessage {
        private String header;
        private String body;
    }
}

JMS的消息类型支持以下几种：

TextMessage：文本消息；
BytesMessage：二进制消息；
MapMessage：包含多个Key-Value对的消息；
ObjectMessage：直接序列化Java对象的消息；
StreamMessage：一个包含基本类型序列的消息。

最常用的是发送基于JSON的文本消息，上述代码通过JmsTemplate创建一个TextMessage并发送到名称为jms/queue/mail的Queue。

注意：Artemis消息服务器默认配置下会自动创建Queue，因此不必手动创建一个名为jms/queue/mail的Queue，但不是所有的消息服务器都会自动创建Queue，生产环境的消息服务器通常会关闭自动创建功能，需要手动创建Queue。

再注意到MailMessage是我们自己定义的一个JavaBean，真正的JMS消息是创建的TextMessage，它的内容是JSON。

当用户注册成功后，我们就调用MessagingService.sendMailMessage()发送一条JMS消息，此代码十分简单，这里不再贴出。

下面我们要详细讨论的是如何处理消息，即编写Consumer。从理论上讲，可以创建另一个Java进程来处理消息，但对于我们这个简单的Web程序来说没有必要，直接在同一个Web应用中接收并处理消息即可。

处理消息的核心代码是编写一个Bean，并在处理方法上标注@JmsListener：

@Component
public class MailMessageListener {
    final Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired ObjectMapper objectMapper;
    @Autowired MailService mailService;

    @JmsListener(destination = "jms/queue/mail", concurrency = "10")
    public void onMailMessageReceived(Message message) throws Exception {
        logger.info("received message: " + message);
        if (message instanceof TextMessage) {
            String text = ((TextMessage) message).getText();
            MailMessage mm = objectMapper.readValue(text, MailMessage.class);
            mailService.sendRegistrationMail(mm);
        } else {
            logger.error("unable to process non-text message!");
        }
    }
}

注意到@JmsListener指定了Queue的名称，因此，凡是发到此Queue的消息都会被这个onMailMessageReceived()方法处理，方法参数是JMS的Message接口，我们通过强制转型为TextMessage并提取JSON，反序列化后获得自定义的JavaBean，也就获得了发送邮件所需的所有信息。

下面问题来了：Spring处理JMS消息的流程是什么？

如果我们直接调用JMS的API来处理消息，那么编写的代码大致如下：

// 创建JMS连接:
Connection connection = connectionFactory.createConnection();
// 创建会话:
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 创建一个Consumer:
MessageConsumer consumer = session.createConsumer(queue);
// 为Consumer指定一个消息处理器:
consumer.setMessageListener(new MessageListener() { 
    public void onMessage(Message message) {
        // 在此处理消息... 
    }
});
// 启动接收消息的循环:
connection.start();

我们自己编写的MailMessageListener.onMailMessageReceived()相当于消息处理器：

consumer.setMessageListener(new MessageListener() { 
    public void onMessage(Message message) {
        mailMessageListener.onMailMessageReceived(message); 
    }
});

所以，Spring根据AppConfig的注解@EnableJms自动扫描带有@JmsListener的Bean方法，并为其创建一个MessageListener把它包装起来。

注意到前面我们还创建了一个JmsListenerContainerFactory的Bean，它的作用就是为每个MessageListener创建MessageConsumer并启动消息接收循环。

再注意到@JmsListener还有一个concurrency参数，10表示可以最多同时并发处理10个消息，5-10表示并发处理的线程可以在5~10之间调整。

因此，Spring在通过MessageListener接收到消息后，并不是直接调用mailMessageListener.onMailMessageReceived()，而是用线程池调用，因此，要时刻牢记，onMailMessageReceived()方法可能被多线程并发执行，一定要保证线程安全。

我们总结一下Spring接收消息的步骤：

通过JmsListenerContainerFactory配合@EnableJms扫描所有@JmsListener方法，自动创建MessageConsumer、MessageListener以及线程池，启动消息循环接收处理消息，最终由我们自己编写的@JmsListener方法处理消息，可能会由多线程同时并发处理。

要验证消息发送和处理，我们注册一个新用户，可以看到如下日志输出：

2020-06-02 08:04:27 INFO  c.i.learnjava.web.UserController - user registered: bob@example.com
2020-06-02 08:04:27 INFO  c.i.l.service.MailMessageListener - received message: ActiveMQMessage[ID:9fc5...]:PERSISTENT/ClientMessageImpl[messageID=983, durable=true, address=jms/queue/mail, ...]]
2020-06-02 08:04:27 INFO  c.i.learnjava.service.MailService - [send mail] sending registration mail to bob@example.com...
2020-06-02 08:04:30 INFO  c.i.learnjava.service.MailService - [send mail] registration mail was sent to bob@example.com.

可见，消息被成功发送到Artemis，然后在很短的时间内被接收处理了。

使用消息服务对发送Email进行改造的好处是，发送Email的能力通常是有限的，通过JMS消息服务，如果短时间内需要给大量用户发送Email，可以先把消息堆积在JMS服务器上慢慢发送，对于批量发送邮件、短信等尤其有用。

小结

ActiveMQ支持基于队列和主题两种模式，即Queue和Topic。

1.基于队列（Queue）的消息系统：在基于队列的消息系统中，生产者将消息发送到队列中，而消费者则从队列中获取消息。队列是一种点对点的通信模型，每个消息只能被一个消费者处理。当有多个消费者连接到同一个队列时，队列将进行负载均衡，以确保每个消费者都能获得相同数量的消息。

2.基于主题（Topic）的消息系统：在基于主题的消息系统中，生产者将消息发送到主题中，而消费者则从主题中订阅消息。主题是一种发布/订阅的通信模型，多个消费者可以同时订阅同一个主题，并且每个消费者都可以接收到相同的消息。当生产者向主题发送一条消息时，所有订阅该主题的消费者都将收到这条消息。

需要注意的是，在使用ActiveMQ时，可以同时支持基于队列和基于主题的消息系统。例如，在订阅主题时可以使用持久订阅来实现基于队列的语义和可靠性，或者使用选择器（Selector）来实现类似于过滤器的功能，以便更精细地控制消息的传递和处理。

总之，在设计和实现基于ActiveMQ的队列和主题的消息系统时，需要根据应用需求和场景选择合适的通信模型，同时还需要考虑消息的路由、传递、确认和处理等方面的问题，以确保消息的可靠传递和高效处理。文章来源地址https://www.toymoban.com/news/detail-455241.html

到了这里，关于ActiveMQ基础学习简单记录的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！