ActiveMQ Artemis中JMX读取字节消息失败的问题排查与解决

问题原因 #

字节消息的CompositeData实例中不存在TEXT_BODY对应的"text"键，直接调用get(CompositeDataConstants.TEXT_BODY)会触发InvalidKeyException。必须先判断消息类型，再选择对应方式读取消息体。

解决步骤 #

1. 识别消息类型：通过CompositeDataConstants.TYPE获取消息类型，ActiveMQ Artemis中主要类型包括TEXT（文本消息）和BYTES（字节消息）。
2. 分类型读取消息体：
   • 文本消息：使用CompositeDataConstants.TEXT_BODY直接获取字符串。
   • 字节消息：使用CompositeDataConstants.BYTES_BODY获取字节数组，再按实际编码转换为字符串。

修改后代码 #

import java.nio.charset.StandardCharsets;
import java.util.Collection;
import javax.management.openmbean.CompositeData;
import javax.management.openmbean.TabularData;
import org.apache.activemq.artemis.api.core.management.CompositeDataConstants;

public static JMXMessage showMessage(QueueControl queueControl) {
    JMXHeaders jmxHeaders;
    JMXMessage jmxMessage;
    String body;
    String replyTo = null;
    String correlationId = null;
    try {
        CompositeData[] messages = queueControl.browse();
        int j = queueLength - i;
        if (j > queueLength - 1) {
            i = 1;
            j = queueLength - i;
        }
        if (j < 0) {
            i = queueLength;
            j = 0;
        }

        // 解析JMS属性
        TabularData stringProps = (TabularData) messages[j].get(CompositeDataConstants.STRING_PROPERTIES);
        for (CompositeData o : (Collection<CompositeData>) stringProps.values()) {
            Object key = o.get("key");
            if ("JMSReplyTo".equals(key)) {
                replyTo = o.get("value").toString();
            }
            if ("JMSCorrelationID".equals(key)) {
                correlationId = o.get("value").toString();
            }
        }

        // 构建消息头
        String messageType = messages[j].get(CompositeDataConstants.TYPE).toString();
        jmxHeaders = new JMXHeaders(messageType,
                messages[j].get(CompositeDataConstants.EXPIRATION).toString(),
                messages[j].get(CompositeDataConstants.DURABLE).toString(),
                replyTo,
                messages[j].get(CompositeDataConstants.MESSAGE_ID).toString(),
                messages[j].get(CompositeDataConstants.TIMESTAMP).toString(),
                correlationId,
                messages[j].get(CompositeDataConstants.PRIORITY).toString());

        // 根据消息类型处理消息体
        if ("TEXT".equals(messageType)) {
            body = (String) messages[j].get(CompositeDataConstants.TEXT_BODY);
        } else if ("BYTES".equals(messageType)) {
            byte[] byteContent = (byte[]) messages[j].get(CompositeDataConstants.BYTES_BODY);
            // 按实际编码转换，示例使用UTF-8
            body = new String(byteContent, StandardCharsets.UTF_8);
        } else {
            // 处理其他未支持的消息类型
            body = "Unsupported message type: " + messageType;
        }

        jmxMessage = new JMXMessage(jmxHeaders, body);
        return jmxMessage;
    } catch (Exception e) {
        throw new RuntimeException(e);
    }
}

注意事项 #

• 消息类型判断：CompositeDataConstants.TYPE返回的字符串与org.apache.activemq.artemis.api.core.Message中的类型常量对应，比如TEXT对应Message.TEXT_TYPE，BYTES对应Message.BYTES_TYPE。
• 字节转字符串编码：必须与消息发送时使用的编码一致，避免乱码，若不确定可尝试常见编码如UTF-8、GBK等。
• 扩展支持：可根据业务需求添加对OBJECT、MAP等其他消息类型的处理逻辑。

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