如何配置Amazon MQ(ActiveMQ+JMS)的最大并发消息处理量

API 探路者

2026-7-5

Amazon MQ（ActiveMQ引擎）JMS消费者串行消费配置问题

问题描述

我正尝试将基于ActiveMQ引擎、采用JMS协议的Amazon MQ集成到应用中，核心需求是让特定流程放缓并在队列中串行执行。我选择JMS集成是因为它能简化监听器实现，在生产者推送消息时异步接收队列消息，但不确定消息是否会被逐个消费。我要求每个消费者一次仅处理一条消息，以此追踪集群节点的最大并发处理消息数，确保单节点仅处理一条消息；至少需要能设置消费者的最大并发消息处理量。

现有消费者代码

@Singleton
public class MmfgActiveMqConsumer {

    Logger LOG = LogManager.getLogger(MmfgActiveMqConsumer.class);

    // Specify the connection parameters.
    private final static String WIRE_LEVEL_ENDPOINT
            = "ssl://b-1234a5b6-78cd-901e-2fgh-3i45j6k178l9-1.mq.us-east-2.amazonaws.com:61617";
    private final static String ACTIVE_MQ_USERNAME = "MyUsername123";
    private final static String ACTIVE_MQ_PASSWORD = "MyPassword456";

    @PostConstruct
    public void init() {
        // Start to listen to ActiveMq Queue
        listenToQueue();
    }

    public void listenToQueue() {
        // Configure connection and session

        new Thread(() -> {
            try {
                final ActiveMQConnectionFactory connectionFactory = createActiveMQConnectionFactory();
                receiveMessage(connectionFactory);
            } catch (Exception e) {
                Thread.currentThread().interrupt();
            }
        }).start();
    }

    private static void receiveMessage(ActiveMQConnectionFactory connectionFactory) throws JMSException {
        // Establish a connection for the consumer.
        // Note: Consumers should not use PooledConnectionFactory.
        final Connection consumerConnection = connectionFactory.createConnection();
        consumerConnection.start();

        // Create a session.
        final Session consumerSession = consumerConnection
                .createSession(false, Session.AUTO_ACKNOWLEDGE);

        // Create a queue named "MyQueue".
        final Destination consumerDestination = consumerSession
                .createQueue("MyQueue");

        // Create a message consumer from the session to the queue.
        final MessageConsumer consumer = consumerSession
                .createConsumer(consumerDestination);

        consumer.setMessageListener(new ActiveMqMessageListener());

        // Thread sleep logic
        //TODO
        // Clean up the consumer.
        consumer.close();
        consumerSession.close();

    }

    private static ActiveMQConnectionFactory createActiveMQConnectionFactory() {
        // Create a connection factory.
        final ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(WIRE_LEVEL_ENDPOINT);

        // Pass the sign-in credentials.
        connectionFactory.setUserName(ACTIVE_MQ_USERNAME);
        connectionFactory.setPassword(ACTIVE_MQ_PASSWORD);
        return connectionFactory;
    }
}

消息监听器代码

public class ActiveMqMessageListener implements MessageListener {

    private Logger LOG = LogManager.getLogger(ActiveMqMessageListener.class);

    @Override
    public void onMessage(Message message) {
        if (message instanceof TextMessage textMessage) {
            try {
                String payload = textMessage.getText();
                LOG.info("Message receive from ActiveMq Queue: " + payload);
            } catch (JMSException e) {
                LOG.error("Error receiving message from ActiveMq Queue: " + e);
            }
        }
    }
}

核心疑问

每次队列收到消息时，onMessage()方法是否会被调用一次，且下一条消息会等待该方法执行完成后再处理？
如何实现单节点仅处理一条消息的需求？Amazon MQ或ActiveMQ是否有特定配置可设置JMS消费者的最大并发消息处理量？

解答

一、`onMessage()`的调用机制

默认情况下，单MessageConsumer实例的onMessage()方法是串行执行的：只有当前onMessage()方法执行完成（包括自动确认消息的逻辑）后，ActiveMQ才会向该消费者推送下一条消息。但你的现有代码存在致命错误：receiveMessage()方法中设置监听器后立刻调用了consumer.close()和consumerSession.close()，这会直接关闭连接，导致监听器根本无法正常接收消息。

二、实现单节点串行消费的具体方案

1. 修复基础代码问题

首先需要修正代码中提前关闭连接/会话的逻辑，确保消费者线程持续运行：

private static void receiveMessage(ActiveMQConnectionFactory connectionFactory) throws JMSException {
    final Connection consumerConnection = connectionFactory.createConnection();
    consumerConnection.start();

    // AUTO_ACKNOWLEDGE模式下，消息会在onMessage执行完成后自动确认
    final Session consumerSession = consumerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    final Destination consumerDestination = consumerSession.createQueue("MyQueue");
    final MessageConsumer consumer = consumerSession.createConsumer(consumerDestination);

    consumer.setMessageListener(new ActiveMqMessageListener());

    // 让当前线程持续阻塞，避免资源被提前关闭
    try {
        Thread.currentThread().join();
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    } finally {
        // 仅在应用停止时优雅关闭资源
        consumer.close();
        consumerSession.close();
        consumerConnection.close();
    }
}

2. 配置预取数保障串行消费

ActiveMQ（包括Amazon MQ）的**预取数（Prefetch Size）**是控制消费者一次性从队列获取消息数量的核心配置。将队列预取数设为1，可确保消费者每次仅获取一条消息，处理完成后才会拉取下一条，从根源上保障串行执行。

代码中配置预取数

private static ActiveMQConnectionFactory createActiveMQConnectionFactory() {
    final ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(WIRE_LEVEL_ENDPOINT);
    connectionFactory.setUserName(ACTIVE_MQ_USERNAME);
    connectionFactory.setPassword(ACTIVE_MQ_PASSWORD);
    
    // 设置队列预取数为1
    ActiveMQPrefetchPolicy prefetchPolicy = new ActiveMQPrefetchPolicy();
    prefetchPolicy.setQueuePrefetch(1);
    connectionFactory.setPrefetchPolicy(prefetchPolicy);
    
    return connectionFactory;
}

通过连接URL配置预取数

也可以直接在连接URL中添加参数：

ssl://b-1234a5b6-78cd-901e-2fgh-3i45j6k178l9-1.mq.us-east-2.amazonaws.com:61617?jms.prefetchPolicy.queuePrefetch=1

3. 自定义最大并发数（扩展方案）

如果后续需要调整单节点的并发处理数（比如从1改为N），可以采用线程池+同步接收的方式，通过线程池大小控制并发量：

// 在MmfgActiveMqConsumer中定义线程池，核心线程数即为最大并发数
private final ExecutorService executor = Executors.newFixedThreadPool(2); // 单节点最大并发2

public void listenToQueue() {
    new Thread(() -> {
        try {
            final ActiveMQConnectionFactory connectionFactory = createActiveMQConnectionFactory();
            receiveMessageWithThreadPool(connectionFactory);
        } catch (Exception e) {
            Thread.currentThread().interrupt();
        }
    }).start();
}

private void receiveMessageWithThreadPool(ActiveMQConnectionFactory connectionFactory) throws JMSException {
    final Connection consumerConnection = connectionFactory.createConnection();
    consumerConnection.start();
    final Session consumerSession = consumerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    final Destination consumerDestination = consumerSession.createQueue("MyQueue");
    final MessageConsumer consumer = consumerSession.createConsumer(consumerDestination);

    // 设置预取数等于线程池大小，避免消息堆积在客户端
    ActiveMQPrefetchPolicy prefetchPolicy = new ActiveMQPrefetchPolicy();
    prefetchPolicy.setQueuePrefetch(2);
    connectionFactory.setPrefetchPolicy(prefetchPolicy);

    while (!Thread.currentThread().isInterrupted()) {
        Message message = consumer.receive();
        if (message != null) {
            // 提交到线程池处理
            executor.submit(() -> {
                try {
                    if (message instanceof TextMessage textMessage) {
                        String payload = textMessage.getText();
                        LOG.info("Message receive from ActiveMq Queue: " + payload);
                    }
                } catch (JMSException e) {
                    LOG.error("Error processing message: ", e);
                }
            });
        }
    }
}

三、Amazon MQ控制台层面的配置

在Amazon MQ控制台中，你也可以对队列进行全局配置：

进入目标队列的配置页面，调整消费者预取数，统一控制该队列所有消费者的预取行为。
若需要全局串行（整个队列仅由一个消费者处理），可以开启队列的Exclusive Consumer（排他消费者），但该配置会导致集群中只有一个消费者能处理队列消息，适用于严格全局串行的场景。

内容的提问来源于stack exchange，提问作者Alessandro Abruzzo

如何配置Amazon MQ(ActiveMQ+JMS)的最大并发消息处理量

问题描述 #

现有消费者代码 #

消息监听器代码 #

核心疑问 #

解答 #

一、onMessage()的调用机制 #

二、实现单节点串行消费的具体方案 #