如何延长ActiveMQ Artemis ClientConsumer.close()的等待超时时间？

针对你遇到的ClientConsumer.close()硬编码10秒超时警告问题，可以通过先停止接收新消息，再等待所有在处理任务完成，最后关闭消费者的方式实现优雅关闭，避免超时警告。以下分两种消费场景给出具体实现：

异步消费场景（使用MessageHandler） #

如果你的消费者是通过setMessageHandler注册异步处理器，可按以下步骤操作：

1. 跟踪活跃任务数：用线程安全的计数器记录正在处理的消息数量，确保能感知所有任务的完成状态。
2. 停止接收新消息：将MessageHandler置为null，阻止消费者接收新消息。
3. 等待任务完成：自定义等待超时（比如50秒，覆盖你的30-40秒处理耗时），直到所有活跃任务结束。
4. 关闭消费者：此时已无在处理任务，调用close()不会触发超时警告。

示例代码：

import java.util.concurrent.atomic.AtomicInteger;
import org.apache.activemq.artemis.api.core.client.ClientConsumer;
import org.apache.activemq.artemis.api.core.client.ClientMessage;
import org.apache.activemq.artemis.api.core.client.MessageHandler;

// 初始化消费者时跟踪活跃任务
AtomicInteger activeProcessingTasks = new AtomicInteger(0);
ClientConsumer consumer = session.createConsumer("your-queue");

MessageHandler messageHandler = new MessageHandler() {
    @Override
    public void onMessage(ClientMessage message) {
        activeProcessingTasks.incrementAndGet();
        try {
            // 执行你的长耗时业务逻辑
            processLongRunningTask(message);
            message.acknowledge();
        } finally {
            activeProcessingTasks.decrementAndGet();
        }
    }
};
consumer.setMessageHandler(messageHandler);

// 优雅关闭方法
public void gracefulShutdown(ClientConsumer consumer) throws InterruptedException {
    // 停止接收新消息
    consumer.setMessageHandler(null);
    
    // 自定义等待超时（50秒，根据实际业务调整）
    long maxWaitTime = 50000;
    long startTime = System.currentTimeMillis();
    
    // 循环等待所有任务完成，或超时退出
    while (activeProcessingTasks.get() > 0 && (System.currentTimeMillis() - startTime) < maxWaitTime) {
        Thread.sleep(100); // 短暂休眠，避免空轮询
    }
    
    // 关闭消费者
    consumer.close();
}

同步消费场景（循环调用receive()） #

如果是同步拉取消息（比如while(true) { ClientMessage msg = consumer.receive(); ... }），可通过以下步骤实现：

1. 设置关闭标志位：用volatile变量标记服务器即将关闭，终止消息拉取循环。
2. 等待当前任务完成：确保正在处理的长耗时任务执行完毕。
3. 关闭消费者：此时无活跃任务，调用close()不会触发超时。

示例代码：

import org.apache.activemq.artemis.api.core.client.ClientConsumer;
import org.apache.activemq.artemis.api.core.client.ClientMessage;

// 关闭标志位，volatile保证多线程可见性
private volatile boolean isShuttingDown = false;

// 消费线程逻辑
public void consumeMessages(ClientConsumer consumer) {
    while (!isShuttingDown) {
        try {
            ClientMessage message = consumer.receive(1000); // 带超时的拉取，避免阻塞
            if (message != null) {
                try {
                    // 执行长耗时业务逻辑
                    processLongRunningTask(message);
                    message.acknowledge();
                } finally {
                    message.release();
                }
            }
        } catch (Exception e) {
            // 处理异常
        }
    }
}

// 优雅关闭方法
public void gracefulShutdown(ClientConsumer consumer) throws InterruptedException {
    // 标记关闭状态，终止消息拉取循环
    isShuttingDown = true;
    
    // 等待当前正在处理的任务完成（可根据实际情况调整等待逻辑，比如用闭锁）
    Thread.sleep(50000); // 直接等待足够长的时间，或用更精确的同步机制
    
    // 关闭消费者
    consumer.close();
}

关键注意事项 #

• 自定义等待超时必须大于你的最长任务处理时间（比如设置为50秒，覆盖30-40秒的耗时）。
• 若使用线程池处理消息，需同时等待线程池中的任务执行完毕，再关闭消费者。
• 避免直接调用close()而不等待任务完成，否则会触发超时警告，甚至可能导致未处理完的消息重复消费（取决于确认机制）。

内容的提问来源于stack exchange，提问作者Cooper Jones