ActiveMQ 高级特性：延迟消息与优先级队列实战（二）

三、优先级队列实战

3.1 优先级队列概念与应用场景

优先级队列是一种特殊的队列，与普通队列按照先进先出（FIFO）的规则不同，优先级队列中的元素按照其优先级进行排序，在消费消息时，高优先级的消息会优先被处理 。这就好比在医院的急诊室，病情危急的患者（高优先级）会比病情较轻的患者（低优先级）优先得到救治。

在任务调度系统中，优先级队列有着广泛的应用。比如在一个分布式计算平台中，不同的任务可能有不同的紧急程度和重要性 。一些实时性要求高的任务，如实时数据分析任务，需要及时处理以提供最新的数据结果，这类任务可以被赋予高优先级；而一些定期执行的批量数据处理任务，如每天凌晨进行的数据备份和统计报表生成任务，实时性要求相对较低，可以设置为低优先级。通过优先级队列，系统可以优先处理高优先级的实时数据分析任务，确保数据的及时性，然后再处理低优先级的批量数据处理任务，合理利用系统资源 。

在电商系统中，订单处理也可以利用优先级队列 。对于一些 VIP 客户的订单，由于他们对商家的贡献较大，为了提供更好的服务体验，这些订单可以被设置为高优先级。而普通客户的订单则设置为低优先级。当订单处理系统从优先级队列中获取订单进行处理时，会优先处理 VIP 客户的订单，优先安排发货、配送等环节，提高 VIP 客户的满意度 。

在物流配送系统中，对于一些加急的快递（如生鲜、药品等时效性强的物品），可以将其配送任务设置为高优先级，优先安排车辆和配送人员进行配送；而普通快递则设置为低优先级，按照正常的配送流程进行处理 。这样可以确保加急快递能够按时送达，满足客户的紧急需求 。

3.2 ActiveMQ 中优先级队列的设置与原理

在 ActiveMQ 中设置优先级队列，需要从两个方面进行操作：一是配置队列属性，使其支持优先级；二是在发送消息时设置消息的优先级 。

在activemq.xml配置文件中，可以通过destinationPolicy节点下的policyEntry来配置队列的优先级属性 。例如，要配置一个名为priorityQueue的队列支持优先级，可添加如下配置：

<destinationPolicy>

<policyMap>

<policyEntries>

<policyEntry queue="priorityQueue" prioritizedMessages="true" />

</policyEntries>

</policyMap>

</destinationPolicy>

上述配置中，prioritizedMessages="true"表示开启该队列的消息优先级功能 。配置完成后，重启 ActiveMQ 服务使配置生效 。

在代码中设置消息的优先级，JMS 规范定义了消息优先级的范围是 0 - 9，其中 0 表示最低优先级，9 表示最高优先级 。在 Java 代码中，使用 ActiveMQ 发送消息时，可以通过以下方式设置消息优先级：

import javax.jms.*;

import org.apache.activemq.ActiveMQConnectionFactory;

public class PriorityMessageProducer {

private static final String BROKER_URL = "tcp://localhost:61616";

private static final String QUEUE_NAME = "priorityQueue";

public static void main(String[] args) throws JMSException {

// 创建连接工厂

ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);

// 创建连接

Connection connection = connectionFactory.createConnection();

// 启动连接

connection.start();

// 创建会话

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

// 创建队列

Destination destination = session.createQueue(QUEUE_NAME);

// 创建消息生产者

MessageProducer producer = session.createProducer(destination);

// 创建高优先级消息

TextMessage highPriorityMessage = session.createTextMessage("高优先级消息");

highPriorityMessage.setJMSPriority(9);

// 发送高优先级消息

producer.send(highPriorityMessage);

// 创建低优先级消息

TextMessage lowPriorityMessage = session.createTextMessage("低优先级消息");

lowPriorityMessage.setJMSPriority(1);

// 发送低优先级消息

producer.send(lowPriorityMessage);

System.out.println("消息已发送");

// 关闭资源

producer.close();

session.close();

connection.close();

}

}

在上述代码中，通过setJMSPriority方法分别为两条消息设置了不同的优先级，一条为 9（最高优先级），一条为 1（较低优先级） 。

ActiveMQ 中优先级队列的原理是，当消费者从队列中获取消息时，ActiveMQ 会根据消息的优先级属性来决定消息的出队顺序 。高优先级的消息会优先被取出，然后才是低优先级的消息 。这样就实现了按照优先级顺序处理消息的功能 。

3.3 实战案例与代码实现

3.3.1 案例背景与需求分析

假设我们正在开发一个分布式任务调度系统，系统中会接收来自不同业务模块的任务，这些任务的紧急程度和重要性各不相同 。例如，业务模块 A 可能会产生一些实时性要求极高的任务，如实时交易数据处理任务，这类任务需要立即得到处理，否则可能会影响交易的正常进行；而业务模块 B 可能会产生一些定期执行的任务，如数据统计分析任务，对实时性要求不高，可以在系统资源空闲时进行处理 。

为了确保系统能够高效地处理这些任务，需要根据任务的优先级进行调度 。高优先级的任务应该优先被分配到计算资源进行处理，低优先级的任务则在高优先级任务处理完成后再进行处理 。因此，我们需要利用 ActiveMQ 的优先级队列特性来实现任务的优先级调度功能 。

3.3.2 代码实现与配置

首先，创建一个 Spring Boot 项目，并在pom.xml文件中添加 ActiveMQ 和 Spring JMS 的依赖：

<dependencies>

<dependency>

<groupId>org.springframework.boot</groupId>

<artifactId>spring-boot-starter-activemq</artifactId>

</dependency>

<dependency>

<groupId>org.springframework.boot</groupId>

<artifactId>spring-boot-starter-jms</artifactId>

</dependency>

</dependencies>

然后，在application.yml文件中配置 ActiveMQ 的连接信息：

spring:

activemq:

broker-url: tcp://localhost:61616

user: admin

password: admin

接下来，在activemq.xml配置文件中配置优先级队列：

<destinationPolicy>

<policyMap>

<policyEntries>

<policyEntry queue="TaskQueue" prioritizedMessages="true" />

</policyEntries>

</policyMap>

</destinationPolicy>

创建任务实体类Task：

import java.io.Serializable;

public class Task implements Serializable {

private static final long serialVersionUID = 1L;

private Long taskId;

private String taskName;

private int priority;

// 其他任务相关属性和getter、setter方法

}

创建消息发送服务类TaskMessageSender：

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.jms.core.JmsTemplate;

import org.springframework.stereotype.Service;

@Service

public class TaskMessageSender {

@Autowired

private JmsTemplate jmsTemplate;

public void sendTaskMessage(Task task) {

jmsTemplate.send("TaskQueue", session -> {

// 创建对象消息，因为要传递任务对象

javax.jms.ObjectMessage objectMessage = session.createObjectMessage(task);

// 设置消息优先级，根据任务的优先级属性

objectMessage.setJMSPriority(task.getPriority());

return objectMessage;

});

}

}

创建消息接收服务类TaskMessageReceiver：

import org.springframework.jms.annotation.JmsListener;

import org.springframework.stereotype.Service;

@Service

public class TaskMessageReceiver {

@JmsListener(destination = "TaskQueue")

public void receiveTaskMessage(Task task) {

System.out.println("接收到任务：" + task.getTaskName() + "，优先级：" + task.getPriority());

// 模拟任务处理逻辑

try {

Thread.sleep(1000);

} catch (InterruptedException e) {

e.printStackTrace();

}

System.out.println("任务 " + task.getTaskName() + " 处理完成");

}

}

在业务逻辑中，调用消息发送服务发送任务消息：

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.web.bind.annotation.PostMapping;

import org.springframework.web.bind.annotation.RequestBody;

import org.springframework.web.bind.annotation.RestController;

@RestController

public class TaskController {

@Autowired

private TaskMessageSender taskMessageSender;

@PostMapping("/tasks")

public String createTask(@RequestBody Task task) {

// 模拟任务创建，设置任务ID和名称

task.setTaskId(1L);

task.setTaskName("任务-" + System.currentTimeMillis());

// 发送任务消息

taskMessageSender.sendTaskMessage(task);

return "任务已创建并发送到队列";

}

}

3.3.3 测试与验证

启动 ActiveMQ 服务和 Spring Boot 应用。使用 Postman 等工具向/tasks接口发送 POST 请求，请求体中包含不同优先级的任务信息，例如：

{

"taskId": 1,

"taskName": "实时交易数据处理任务",

"priority": 9

}

{

"taskId": 2,

"taskName": "数据统计分析任务",

"priority": 1

}

发送多个不同优先级的任务请求后，查看控制台输出 。可以看到，高优先级的任务会优先被接收和处理，低优先级的任务在高优先级任务处理完成后才会被处理 。通过这种方式，可以验证消息是否按照优先级顺序被消费，从而验证优先级队列在任务调度系统中的功能是否正常实现 。

四、高级应用与优化

4.1 延迟消息与优先级队列的结合使用

在复杂的业务场景中，将延迟消息和优先级队列结合起来使用，可以更灵活地满足业务需求 。

以电商系统的订单处理为例，假设系统中有普通订单、VIP 订单以及促销活动订单。对于普通订单，在用户下单后，如果 30 分钟内未支付，系统自动取消订单，这可以通过延迟消息来实现；对于 VIP 订单，为了提供更好的服务体验，不仅要保证其优先处理，还可以设置较短的支付超时时间（如 15 分钟），这就需要结合优先级队列和延迟消息；对于促销活动订单，由于活动期间订单量较大，为了确保活动的顺利进行，可能需要对这些订单设置较高的优先级，并且根据活动规则设置不同的延迟处理时间 。

实现思路如下：

1. 配置队列：在activemq.xml文件中，配置订单队列使其支持优先级，例如：

<destinationPolicy>

<policyMap>

<policyEntries>

<policyEntry queue="OrderQueue" prioritizedMessages="true" />

</policyEntries>

</policyMap>

</destinationPolicy>

1. 发送消息：在发送订单消息时，根据订单类型设置消息的优先级和延迟时间。例如，使用 Java 代码发送订单消息：

import org.apache.activemq.ScheduledMessage;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.jms.core.JmsTemplate;

import org.springframework.stereotype.Service;

@Service

public class OrderMessageSender {

@Autowired

private JmsTemplate jmsTemplate;

public void sendOrderMessage(Order order, int priority, long delay) {

jmsTemplate.send("OrderQueue", session -> {

// 创建对象消息，因为要传递订单对象

javax.jms.ObjectMessage objectMessage = session.createObjectMessage(order);

// 设置消息优先级

objectMessage.setJMSPriority(priority);

// 设置延迟发送时间，单位毫秒

objectMessage.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, delay);

return objectMessage;

});

}

}

在上述代码中，sendOrderMessage方法接收订单对象、优先级和延迟时间作为参数。根据订单类型，如 VIP 订单可以设置优先级为 9，延迟时间为 15 * 60 * 1000（15 分钟）；普通订单设置优先级为 5，延迟时间为 30 * 60 * 1000（30 分钟） 。

3. 接收消息：消费者从队列中接收消息时，ActiveMQ 会根据消息的优先级和延迟时间，优先处理高优先级且延迟时间已到的消息 。消费者代码与前面介绍的类似，通过@JmsListener注解监听队列并处理消息：

import org.springframework.jms.annotation.JmsListener;

import org.springframework.stereotype.Service;

@Service

public class OrderMessageReceiver {

@JmsListener(destination = "OrderQueue")

public void receiveOrderMessage(Order order) {

// 处理订单逻辑

}

}

4.2 性能优化与注意事项

在使用延迟消息和优先级队列时，可能会出现一些性能问题，需要采取相应的优化措施和注意事项 。

性能问题：

• 消息堆积：如果生产者发送消息的速度远大于消费者处理消息的速度，无论是延迟消息还是普通消息，都会导致消息在队列中堆积，占用大量的内存和磁盘空间，影响系统性能 。特别是在高并发场景下，若优先级队列中高优先级消息持续产生，低优先级消息可能会长时间得不到处理，进一步加剧消息堆积 。

• 资源消耗：ActiveMQ 在处理延迟消息时，需要维护一个调度任务列表，随着延迟消息数量的增加，调度任务的管理和执行会消耗更多的 CPU 和内存资源 。对于优先级队列，在排序和调度消息时也会增加系统开销 。

优化建议：

• 合理设置队列参数：通过调整队列的concurrentConsumers（并发消费者数量）和prefetch（预取数量）等参数，可以提高消息的消费速度 。例如，在activemq.xml文件中配置：

<destinationPolicy>

<policyMap>

<policyEntries>

<policyEntry queue="OrderQueue" concurrentConsumers="10" prefetch="50" />

</policyEntries>

</policyMap>

</destinationPolicy>

上述配置中，concurrentConsumers="10"表示允许同时有 10 个消费者处理该队列的消息，prefetch="50"表示每个消费者一次从队列中预取 50 条消息，具体数值可根据实际业务量和系统性能进行调整 。

• 使用异步处理：在消费者端，对于消息的处理逻辑可以采用异步方式，如使用线程池来处理消息，避免单个消息处理时间过长导致其他消息等待 。例如：

import org.springframework.jms.annotation.JmsListener;

import org.springframework.stereotype.Service;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

@Service

public class OrderMessageReceiver {

private static final ExecutorService executorService = Executors.newFixedThreadPool(5);

@JmsListener(destination = "OrderQueue")

public void receiveOrderMessage(Order order) {

executorService.submit(() -> {

// 处理订单逻辑

});

}

}

在上述代码中，创建了一个固定大小为 5 的线程池，当接收到消息时，将消息处理任务提交到线程池中异步执行 。

• 优化消息存储：对于持久化的消息，选择合适的持久化策略和存储介质可以提高性能 。例如，使用 KahaDB 存储时，可以通过调整数据文件大小等参数来优化存储性能；使用数据库持久化时，优化数据库表结构和索引，提高数据读写速度 。

注意事项：

• 优先级设置的合理性：在设置消息优先级时，要根据业务的实际需求合理分配优先级，避免出现优先级设置混乱导致重要消息得不到及时处理的情况 。同时，要注意不同队列之间的优先级是相互独立的，不要混淆 。

• 延迟时间的准确性：设置延迟消息的延迟时间时，要考虑系统的时间精度和误差，确保延迟时间能够满足业务需求 。如果延迟时间设置过短，可能无法达到预期的延迟效果；设置过长，则可能导致业务处理不及时 。

五、总结与展望

在分布式系统开发中，ActiveMQ 的延迟消息和优先级队列特性为解决复杂业务场景下的消息处理问题提供了强大的支持 。通过本次实战，我们深入了解了这两个特性的概念、实现方式以及在实际项目中的应用 。

在延迟消息方面，我们掌握了使用 TimeToLive 属性和 Scheduled Message 机制两种实现方式。TimeToLive 属性实现简单，适用于对延迟时间精度要求不高的场景；Scheduled Message 机制则更加灵活，能够满足复杂的定时任务需求 。通过电商系统订单支付超时取消的案例，我们看到了延迟消息在实际业务中的具体应用，有效地提高了系统的自动化处理能力和业务的准确性 。

对于优先级队列，我们学会了在 ActiveMQ 中配置队列支持优先级，并在发送消息时设置消息的优先级 。通过分布式任务调度系统的案例，验证了优先级队列能够根据任务的优先级进行合理调度，确保高优先级任务优先得到处理，提高了系统的整体效率和性能 。

将延迟消息和优先级队列结合使用，进一步拓展了 ActiveMQ 在复杂业务场景中的应用。通过电商系统订单处理的案例，展示了如何根据订单类型设置不同的优先级和延迟时间，实现了更加灵活和高效的业务处理逻辑 。

在未来，随着分布式系统的不断发展和业务需求的日益复杂，消息队列技术也将不断演进 。ActiveMQ 作为一款成熟的消息中间件，有望在性能优化、功能扩展等方面持续发展，更好地满足开发者的需求 。例如，在性能优化方面，可能会进一步改进消息的存储和调度算法，提高消息处理的速度和吞吐量；在功能扩展方面，可能会支持更多的消息协议和高级特性，如与云原生技术的深度融合，为分布式系统的开发提供更强大的支持 。同时，我们也期待 ActiveMQ 能够在更多的领域和场景中得到应用，为企业的数字化转型和业务创新提供有力的支撑 。