ActiveMQ Artemis ClientMessage消息确认机制异常问题咨询

问题背景 #

使用ActiveMQ Artemis核心API的ClientMessage作为消息容器时，调用acknowledge()方法无法确认消息（消息仍留在队列中），但individualAcknowledge()方法可正常工作；而使用JMS规范的javax.jms.Message时，acknowledge()却能正常生效。相关处理代码如下：

@Override
public void onMessage(ClientMessage message) {
    try {
        // acknowledge() 方法不生效，消息仍在队列中
        // message.acknowledge();
        message.individualAcknowledge();
    } catch (ActiveMQException e) {
        log.error("message acknowledge error: ", e);
    }
}

核心源码对比分析 #

ClientMessage的acknowledge()逻辑 #

public void acknowledge(ClientMessage message) throws ActiveMQException {
    ClientMessageInternal cmi = (ClientMessageInternal)message;
    if (this.ackIndividually) {
        this.individualAcknowledge(message);
    } else {
        this.ackBytes += message.getEncodeSize();
        if (logger.isTraceEnabled()) {
            logger.trace(this + "::acknowledge ackBytes=" + this.ackBytes + " and ackBatchSize=" + this.ackBatchSize + ", encodeSize=" + message.getEncodeSize());
        }

        if (this.ackBytes >= this.ackBatchSize) {
            if (logger.isTraceEnabled()) {
                logger.trace(this + ":: acknowledge acking " + cmi);
            }

            this.doAck(cmi);
        } else {
            if (logger.isTraceEnabled()) {
                logger.trace(this + ":: acknowledge setting lastAckedMessage = " + cmi);
            }

            this.lastAckedMessage = cmi;
        }
    }
}

javax.jms.Message的acknowledge()逻辑 #

public void acknowledge() throws JMSException {
    if (this.session != null) {
        try {
            if (this.session.isClosed()) {
                throw ActiveMQClientMessageBundle.BUNDLE.sessionClosed();
            }

            if (this.individualAck) {
                this.message.individualAcknowledge();
            }

            if (this.clientAck || this.individualAck) {
                this.session.commit(this.session.isBlockOnAcknowledge());
            }
        } catch (ActiveMQException var2) {
            throw JMSExceptionHelper.convertFromActiveMQException(var2);
        }
    }
}

从源码逻辑可明确：

• JMS规范的acknowledge()最终会调用session.commit()，强制触发确认操作，不受批量阈值限制。
• 核心API的ClientMessage.acknowledge()采用批量确认机制：仅当累计确认的消息字节数达到ackBatchSize阈值时，才会调用doAck()向Broker发送确认指令；若未达标，仅将消息暂存到lastAckedMessage，不会立即触发Broker侧的消息确认。

问题解答 #

1. 单条消息未达批量阈值时，lastAckedMessage的处理方式 #

当单条消息字节数远小于ackBatchSize时，lastAckedMessage会暂存该消息，但不会立即向Broker发送确认指令。此时Broker未收到确认信号，消息会一直处于未确认状态并留在队列中。

2. 如何确认这条消息？ #

有三种可行方式：

• 手动提交会话：调用核心API的ClientSession.commit()，强制将暂存的lastAckedMessage及累计ackBytes对应的消息全部确认，逻辑与JMS API一致。
• 调整批量阈值：将ackBatchSize设置为极小值（如1字节），使单条消息即可触发doAck()执行确认。
• 继续使用individualAcknowledge()：该方法绕过批量机制，直接调用doAck()发送单条消息的确认指令，这也是你当前验证有效的方案。

3. acknowledge()失效的原因 #

核心API的acknowledge()默认采用批量确认策略，你配置的autoCommitSends和autoCommitAcks仅对JMS规范的API生效，对核心API的批量确认逻辑无影响。只有当累计确认字节数达标，或手动调用session.commit()时，才会真正向Broker发送确认指令。而JMS的acknowledge()因内部强制触发了session.commit()，所以不受批量阈值限制，能立即确认消息。

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