一、故障背景与现象描述

在一个典型的电商大促场景中，订单系统每秒产生数千条消息投递到 ActiveMQ，消费端却在几小时内完全瘫痪。监控面板上，积压消息数从两位数一路飙到百万级，订单状态迟迟不更新，用户投诉电话被打爆。这不是个例，而是消息队列在生产环境中最容易出问题的环节之一。

消息堆积说白了就是生产者生产消息的速度远快于消费者消费的速度，或者消费者干脆卡死了，消息只进不出，队列越积越多，最终可能导致内存溢出、磁盘写满，整个服务雪崩。

ActiveMQ 作为 Apache 基金会下最老牌的消息队列，在很多老项目里仍在运行。它支持多种协议，部署简单，但在高吞吐场景下，不做好调优就容易踩坑。

二、消息堆积原因深度分析

2.1 生产者发送过快

生产者端如果没有限流，或者批量推送时一次性把几十万条消息塞进去，ActiveMQ 的 Broker 端内存压力会急剧上升。尤其是使用了同步发送模式时，队列堆积会直接影响生产者的吞吐量。

技术栈：Java + Spring Boot + ActiveMQ

/**
 * 生产者示例 - 演示批量消息发送可能造成的堆积
 * 技术栈：Java + Spring Boot + ActiveMQ
 */
@Component
public class OrderMessageProducer {

    @Autowired
    private JmsTemplate jmsTemplate;

    /**
     * 批量发送订单消息
     * 注意：如果订单量大，一次性发送会造成 Broker 端压力骤增
     */
    public void batchSendOrders(List<Order> orders) {
        for (Order order : orders) {
            // 将订单对象转为 JSON 字符串消息
            String jsonMessage = JsonUtils.toJson(order);
            // 同步发送到订单队列
            jmsTemplate.convertAndSend("order.queue", jsonMessage);
        }
    }

    /**
     * 带限流的发送方式 - 推荐在高并发场景使用
     * @param orders 订单列表
     * @param batchSize 每批处理数量
     * @param sleepMs 每批之间的休眠时间（毫秒）
     */
    public void batchSendWithRateLimit(List<Order> orders,
                                        int batchSize,
                                        long sleepMs) {
        int count = 0;
        for (Order order : orders) {
            String jsonMessage = JsonUtils.toJson(order);
            jmsTemplate.convertAndSend("order.queue", jsonMessage);
            count++;
            // 每发送一批就暂停一下，给消费者喘息的机会
            if (count % batchSize == 0) {
                try {
                    Thread.sleep(sleepMs);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }
    }
}

2.2 消费者处理过慢

消费者端如果出现了数据库锁、网络超时、第三方接口响应慢等问题，消息消费速度就会下降。在 Spring Boot 里使用 @JmsListener 注解监听队列时，如果处理方法中有耗时操作又没有做异步处理，积压就会快速累积。

/**
 * 消费者示例 - 展示消费慢导致堆积的典型场景
 * 技术栈：Java + Spring Boot + ActiveMQ
 */
@Component
public class OrderMessageConsumer {

    @Autowired
    private OrderService orderService;

    /**
     * 监听订单队列并处理消息
     * 注意：如果 processOrder 方法中调用了外部接口或慢查询，
     * 消费速度会跟不上生产速度，导致消息堆积
     */
    @JmsListener(destination = "order.queue",
                 concurrency = "1")  // 并发数为1，单线程消费
    public void listenOrder(String message) {
        Order order = JsonUtils.fromJson(message, Order.class);
        // 这个方法如果耗时过长（比如调用了第三方物流接口），
        // 就会成为堆积的瓶颈
        orderService.processOrder(order);
    }
}

2.3 消费者宕机或网络中断

消费者服务如果因为内存溢出、Full GC、网络抖动等原因短暂不可用，消息会在队列里持续堆积。如果队列没有设置持久化策略，消费者恢复后可能面临数据丢失的风险。

2.4 Broker 端配置不当

ActiveMQ 默认的 Broker 配置在低负载下没问题，但在高并发场景下，如果没有合理设置持久化策略、内存阈值、磁盘阈值，堆积到一定程度就会触发故障转移或者直接 OOM。

三、生产故障排查全流程

3.1 第一步：确认堆积规模

登录 ActiveMQ 的 Web 管理控制台，查看各队列的积压消息数。默认访问地址是 http://localhost:81611，进去后重点关注 Pending（未确认消息数）和 Message Count（队列消息总数）。

也可以使用命令行工具快速查询：

# 通过 ActiveMQ 的 CLI 工具查询队列积压情况
# 需要安装 activemq 命令行客户端
# -u 和 -p 分别指定用户名和密码，默认是 admin/admin
$ amq broker --connect tcp://broker-host:61616 --user admin --password admin
> queue list
> queue stats order.queue

3.2 第二步：定位堆积原因

判断堆积是生产太快还是消费太慢，可以通过以下几个维度排查：

查看消费者日志：如果消费者日志中频繁出现异常堆栈，说明处理过程中有报错导致消息无法正常 ACK。

# 查看消费者服务日志，搜索异常信息
# -C 表示在文件间循环搜索，类似 tail -f
# --grep 过滤包含 ERROR 或 Exception 的行
$ tail -C -n 500 /var/log/app/order-consumer.log | grep -E "ERROR|Exception"

检查消费者服务健康状态：确认消费者进程是否存活，内存和 CPU 使用是否正常。

# 检查消费者服务进程
$ ps -ef | grep order-consumer

# 查看消费者服务的内存使用情况
$ jstat -gcutil <消费者进程PID> 1000 5

# 查看消费者服务线程状态，确认是否卡在某个方法上
$ jstack <消费者进程PID> > thread_dump.txt
# 查看 dump 文件中的 BLOCKED 和 WAITING 线程
$ grep -A 20 "BLOCKED\|WAITING" thread_dump.txt

检查 ActiveMQ Broker 端状态：

# 查看 Broker 端内存和磁盘使用情况
$ amq broker --connect tcp://broker-host:61616 --user admin --password admin
> broker stats

# 查看是否有消费者断开连接的情况
> consumers list

# 查看 Producer 的发送速率
> producers stats

3.3 第三步：紧急处理措施

如果堆积已经影响到业务，需要先采取紧急措施控制影响面：

方案一：临时提升消费者并发数

/**
 * 动态调整消费者并发数 - 通过配置修改
 * 技术栈：Java + Spring Boot + ActiveMQ
 */
@Configuration
public class JmsConfig {

    @Value("${jms.listener.concurrency:5}")
    private String concurrency;

    @Bean
    public DefaultJmsListenerContainerFactory jmsListenerContainerFactory(
            ConnectionFactory connectionFactory) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        // 设置并发消费者数量，范围 1-10
        factory.setConcurrency(concurrency);
        factory.setSessionTransacted(false);
        return factory;
    }
}

对应的配置文件：

# application.yml 中的配置
# 并发消费者数量，堆积时可临时调高
jms.listener.concurrency=10
# 是否启用集群模式（多实例部署时设为 true）
jms.listener.cluster=true
# 消息接收超时时间（毫秒）
spring.jms.listener.acknowledge-timeout=60000

方案二：临时关闭或降级非核心业务的生产者

如果订单队列堆积严重，但优惠券、通知等队列堆积较轻，可以暂时将部分非核心消息改为本地队列存储或延时处理，减轻 ActiveMQ 的压力。

方案三：消息批量丢弃（谨慎使用）

对于已经过期且无价值的消息，可以通过 ActiveMQ 的管理接口直接清空队列。这一步操作要非常谨慎，确保不会丢弃需要处理的有效消息。

# 通过 Admin API 清空指定队列（慎用！）
# 这个操作会丢失所有未消费的消息，确认无业务影响后再执行
curl -X POST "http://broker-host:8161/admin/queues/order.queue?action=clear" \
  -u admin:admin

# 清空前先导出备份（如有持久化策略）
# 将 KahaDB 目录备份到安全位置
$ cp -r /opt/activemq/data/ /opt/activemq/data_backup_$(date +%Y%m%d_%H%M%S)

3.4 第四步：恢复后验证

堆积处理完成后，需要观察一段时间确认系统恢复到稳定状态。重点关注以下指标：

  • 队列 Pending 数是否持续下降
  • 消费者 CPU 和内存是否稳定
  • 消息处理延迟是否恢复到正常范围
  • 下游业务系统数据是否一致
# 持续监控队列积压情况，每 5 秒刷新一次
$ watch -n 5 "amq broker --connect tcp://broker-host:61616 --user admin --password admin <<< 'queue stats order.queue'"

# 检查消费者消息处理延迟
$ curl -s "http://consumer-host:8080/actuator/metrics/jms.listener.processing.time" | jq .

四、性能调优实战方案

4.1 Broker 端调优

ActiveMQ 的 activemq.xml 配置文件是调优的核心。针对消息堆积场景，需要重点调整以下几个参数：

<!-- activemq.xml 配置示例 - 针对高吞吐场景的调优配置 -->
<!-- 技术栈：Java + ActiveMQ Broker 5.17+ -->

<broker xmlns="http://activemq.apache.org/schema/core"
        brokerName="production-broker"
        persistent="true"
        dataDirectory="/opt/activemq/data">

    <!-- 内存管理策略 -->
    <memoryPolicy>
        <!-- 每个目的地最多占用 Broker 可用内存的比例 -->
        <tempUsage percentageOfJvmHeap="40"/>
        <persistenceLimit>100mb</persistenceLimit>
    </memoryPolicy>

    <!-- 持久化策略：使用 KahaDB 提升读写性能 -->
    <persistenceAdapter>
        <kahaDB directory="/opt/activemq/data/compacted"
                journalMaxFileLength="100mb"
                checkpointInterval="5000"
                enableJournalDiskSync="false"
                enableJournalDiskSynchronousDelete="true"/>
    </persistenceAdapter>

    <!-- 网络运输层：提升传输性能 -->
    <transportConnectors>
        <transportConnector name="openwire"
                           uri="tcp://0.0.0.0:61616?wireFormat.optimizedAcknowledge=true&wireFormat.preserveConsumerOrder=true"/>
        <transportConnector name="stomp"
                           uri="stomp://0.0.0.0:61613"/>
    </transportConnectors>

    <!-- 虚拟目的地：解决消费者组共享队列的问题 -->
    <virtualDestinations>
        <virtualTopic name="topic.order"
                      physicalDestinations="queue.order">
        </virtualTopic>
    </virtualDestinations>

    <!-- 插件配置 -->
    <plugins>
        <!-- 死信队列：处理无法消费的消息 -->
        <deadLetterStrategy deleteAllMessagesOnStartup="false">
            <queue physicalName="DLQ.ORDER_QUEUE"/>
        </deadLetterStrategy>

        <!-- 重投递策略：失败消息的重新投递控制 -->
        <redeliveryPolicy maximumRedeliveries="3"
                          backOffMultiplier="2"
                          initialDelay="1000"
                          useExponentialBackOff="true"/>

        <!-- 消息追踪：用于问题排查 -->
        <tracePlugin trace="true" traceAll="false"
                     traceExpiry="true" traceAcknowledge="true"/>
    </plugins>
</broker>

4.2 消费者端调优

/**
 * 高性能消费者配置 - 针对消息堆积场景的优化方案
 * 技术栈：Java + Spring Boot + ActiveMQ
 */
@Configuration
public class HighPerformanceJmsConfig {

    /**
     * 配置连接工厂，启用连接池
     * 使用 ActiveMQ 的 PooledConnectionFactory 提升连接复用率
     */
    @Bean
    public ActiveMQConnectionFactory connectionFactory(
            @Value("${spring.activemq.broker-url}") String brokerUrl) {

        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory();
        factory.setBrokerURL(brokerUrl);
        // 连接超时时间（毫秒）
        factory.setConnectionTimeout(30000);
        // 接收超时时间（毫秒）
        factory.setReceiveTimeout(30000);
        // 是否支持异步发送，提升生产者发送效率
        factory.setAsyncSend(true);
        return factory;
    }

    /**
     * 配置连接池，避免频繁创建和销毁连接
     * 连接池大小应与消费者并发数匹配
     */
    @Bean
    public ActiveMQConnection activeMQConnection(
            ActiveMQConnectionFactory factory) {
        ActiveMQConnection connection = new ActiveMQConnection(factory);
        // 设置连接池大小
        connection.setMaxConnections(20);
        // 连接空闲超时时间（秒）
        connection.setExpiryTimeout(60);
        return connection;
    }

    /**
     * 配置事务管理器
     */
    @Bean
    public JmsTransactionManager transactionManager(
            ConnectionFactory connectionFactory) {
        return new JmsTransactionManager(connectionFactory);
    }

    /**
     * 配置消费者容器工厂
     */
    @Bean
    public DefaultJmsListenerContainerFactory jmsListenerContainerFactory(
            ConnectionFactory connectionFactory,
            JmsTransactionManager transactionManager) {

        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setTransactionManager(transactionManager);
        // 并发消费者数量，根据 CPU 核心数合理设置
        // 建议公式：核数 × 2（纯 CPU 密集型）或 核数 × 4（IO 密集型）
        factory.setConcurrency("5-15");
        // 批处理大小，一次从队列拉取的消息数
        factory.setBatchSize(100);
        // 是否开启会话事务
        factory.setSessionTransacted(true);
        // 消费者空闲等待时间（毫秒）
        factory.setReceiveTimeout(30000);
        return factory;
    }
}

4.3 消费者业务逻辑优化

/**
 * 优化后的消费者实现 - 避免单条慢消息拖慢整体消费速度
 * 技术栈：Java + Spring Boot + ActiveMQ
 */
@Component
public class OptimizedOrderConsumer {

    @Autowired
    private OrderService orderService;

    @Autowired
    private AsyncThreadPool asyncThreadPool;

    @Autowired
    private MessageBuffer messageBuffer;

    /**
     * 优化方案一：异步处理耗时操作
     * 将慢操作提交到线程池，消费者快速 ACK 消息
     * 注意：需要确保异步任务的可靠性，建议使用内部消息队列或数据库记录
     */
    @JmsListener(destination = "order.queue",
                 concurrency = "5-15",
                 ackMode = "AUTO")
    public void consumeFast(String message) {
        Order order = JsonUtils.fromJson(message, Order.class);

        // 将耗时处理异步化
        CompletableFuture.runAsync(() -> {
            try {
                orderService.processOrder(order);
            } catch (Exception e) {
                // 记录失败日志，触发告警
                log.error("Order processing failed for order: {}",
                          order.getOrderId(), e);
                // 可选：将失败消息发送到重试队列
                messageBuffer.put("retry.order.queue", message);
            }
        }, asyncThreadPool.getExecutor());
    }

    /**
     * 优化方案二：批量消费
     * 一次性处理多条消息，减少网络往返次数
     */
    @JmsListener(destination = "order.queue",
                 concurrency = "3-10",
                 batch = "100",
                 acknowledge = "AUTO")
    public void consumeBatch(String[] messages) {
        List<Order> orders = Arrays.stream(messages)
                .map(msg -> JsonUtils.fromJson(msg, Order.class))
                .collect(Collectors.toList());

        // 批量处理，一次性提交数据库事务
        orderService.batchProcessOrders(orders);
    }

    /**
     * 优化方案三：消息过滤 + 前置校验
     * 在消费入口做快速失败，避免无效消息进入处理流程
     */
    @JmsListener(destination = "order.queue",
                 concurrency = "5-15")
    public void consumeWithValidation(String message) {
        Order order = JsonUtils.fromJson(message, Order.class);

        // 前置校验：快速过滤无效消息
        if (order == null || order.getOrderId() == null) {
            log.warn("Received invalid message, skipping: {}", message);
            return;
        }

        if (order.getStatus() == OrderStatus.CANCELLED) {
            log.info("Order {} is cancelled, skip processing", order.getOrderId());
            return;
        }

        // 检查是否已经处理过（幂等性校验）
        if (orderService.isProcessed(order.getOrderId())) {
            log.debug("Order {} already processed, skip", order.getOrderId());
            return;
        }

        // 正常业务处理
        orderService.processOrder(order);
    }
}

4.4 生产者端调优

/**
 * 生产者配置优化 - 控制发送速率，避免压垮 Broker
 * 技术栈：Java + Spring Boot + ActiveMQ
 */
@Configuration
public class OptimizedProducerConfig {

    @Bean
    public JmsTemplate jmsTemplate(ConnectionFactory connectionFactory) {
        JmsTemplate template = new JmsTemplate(connectionFactory);
        // 消息发送超时时间（毫秒）
        template.setDeliveryTimeout(10000);
        // 消息时间戳
        template.setDeliveryMode(DeliveryMode.PERSISTENT);
        // 是否启用批量发送
        template.setBatchEnabled(true);
        return template;
    }
}

@Component
public class RateLimitedProducer {

    @Autowired
    private JmsTemplate jmsTemplate;

    // 使用 Guava 的 RateLimiter 控制发送速率
    private final RateLimiter rateLimiter = RateLimiter.create(5000.0);

    /**
     * 带限流的消息发送
     * 每秒最多发送 5000 条消息，超出则阻塞等待
     */
    public void sendWithRateLimit(String destination, String message) {
        // 获取许可，如果速率超限会阻塞
        rateLimiter.acquire();

        MessageProperties postProperties = new MessageProperties();
        postProperties.setDeliveryMode(DeliveryMode.PERSISTENT);
        // 设置消息过期时间，避免过期消息堆积
        postProperties.setTimeToLive(300000); // 5 分钟

        jmsTemplate.send(destination, session -> {
            TextMessage textMessage = session.createTextMessage(message);
            // 设置消息 ID 方便追踪
            textMessage.setJMSMessageID("MSG_" + UUID.randomUUID());
            return textMessage;
        });
    }
}

五、应用场景分析

5.1 电商订单系统

订单系统是高并发消息场景的典型代表。用户在下单过程中，需要解耦订单创建、库存扣减、支付通知、发货等多个环节。ActiveMQ 在这种场景下负责削峰填谷，把瞬时洪峰打散到后续处理中。但正因为流量大，消息堆积的风险也高。

5.2 日志采集系统

日志采集系统通常由 Logstash 或类似工具通过 ActiveMQ 转发日志消息。在服务器数量多、日志量大时，消费端如果处理不过来，日志会大量积压，影响实时监控告警的时效性。

5.3 异步通知系统

短信、邮件、推送等通知服务通常不要求实时性，适合通过消息队列异步处理。但这类场景下如果消费者长期宕机，消息堆积会导致大量用户收不到通知，影响用户体验。

六、技术优缺点对比

6.1 ActiveMQ 的优势

ActiveMQ 是老牌的消息中间件，最大的优势在于协议支持丰富，支持 JMS 标准规范，兼容性好。对于使用 Java 技术栈的系统来说，集成成本很低，Spring Boot 几行配置就能跑起来。同时，它的集群模式成熟稳定，支持主备和广播两种集群方式。

6.2 ActiveMQ 的劣势

ActiveMQ 在高吞吐场景下的性能不如 Kafka 和 RocketMQ。特别是 ActiveMQ Classic（5.x 版本）在百万级消息堆积时，磁盘 IO 会成为明显瓶颈。另外，ActiveMQ 的监控工具相对简陋，缺乏像 Kafka 那样的生态工具链。

6.3 与其他消息中间件的对比

特性 ActiveMQ Classic RabbitMQ RocketMQ Kafka
吞吐量 万级/秒 十万级/秒 十万级/秒 百万级/秒
消息堆积能力 中等 中等 强 极强
协议支持 丰富 丰富 较少 自有协议
学习成本 低 中 中 高
运维复杂度 低 中 中 高

七、注意事项与最佳实践

7.1 消息持久化与可靠性

对于重要业务消息，必须开启持久化。但持久化会带来额外的磁盘 IO 开销，需要在可靠性和性能之间做权衡。建议对非核心消息使用非持久化模式，核心业务消息使用持久化模式。

7.2 消费者幂等性

消息队列不保证消息只投递一次（除非使用严格的事务消息），所以消费者端必须做好幂等性处理。可以通过消息 ID 去重、业务状态机校验等方式实现。

7.3 死信队列处理

被多次消费失败的消息会进入死信队列（DLQ）。如果不及时处理，死信队列本身也会堆积。建议定期扫描死信队列，分析失败原因，修复后重新投递。

7.4 监控告警体系

建立完整的监控告警体系是防止消息堆积演变为生产故障的关键。建议监控以下指标：

  • 队列消息积压数
  • 消费者消费速率
  • 生产者发送速率
  • 消息处理延迟（P99）
  • Broker 内存和磁盘使用率
# 通过 Prometheus 采集 ActiveMQ 指标
# 在 activemq.xml 中启用 Prometheus 导出
# 然后配置 Prometheus 定期拉取指标

# 查看 Broker 关键指标
curl -s "http://broker-host:8161/admin/api/broker.json" | jq '.memoryUsage'
curl -s "http://broker-host:8161/admin/api/broker.json" | jq '.storeUsage'
curl -s "http://broker-host:8161/admin/api/broker.json" | jq '.tempUsage'

# 查看队列级别的指标
curl -s "http://broker-host:8161/admin/api/queues/order.queue" | jq '{
    messages: .messages,
    pending: .pending,
    consumers: .consumers,
    totalDequeueCount: .totalDequeueCount,
    totalEnqueueCount: .totalEnqueueCount
}'

7.5 灰度发布策略

在消息队列相关的代码变更上线时，务必采用灰度策略。可以先在测试环境验证堆积处理能力，再选择非核心队列灰度上线，观察稳定后再全量推广。

八、文章总结

消息堆积是消息队列在生产环境中最常见的故障类型之一。排查和解决消息堆积，需要从生产端、Broker 端、消费端三个维度同时入手，而不是只盯着某一个环节优化。

核心思路可以归纳为以下几点：

第一，快速定位堆积原因。是通过生产者发送速率监控、消费者日志分析、Broker 状态检查三步走快速判断问题出在哪个环节。

第二，紧急止血措施。在排查过程中如果业务受到影响，需要先采取提升消费者并发数、降低非核心生产者速率、必要时丢弃过期消息等措施控制影响面。

第三，性能调优。Broker 端的内存策略、持久化策略、队列策略需要针对实际业务场景调优；消费者端的并发数、批处理大小、异步处理机制都需要合理配置；生产者端需要加入限流机制，避免瞬时压力过大。

第四，建立长效监控体系。监控消息积压量、消费速率、处理延迟等核心指标，设置合理的告警阈值，做到问题在堆积早期就被发现和处理，而不是等到堆积到百万级才被动救火。

消息队列是分布式系统的核心基础设施，做好消息堆积的预防和应对方案，是保障系统稳定运行的必修课。