如何对ActiveMQ Artemis特定消费者的JMS消息进行日志追踪

一、通过日志框架实现精准过滤 #

利用org.apache.activemq.artemis.jms日志器，结合Log4j2、Logback等日志框架的过滤规则，只输出符合条件的日志，规避全量追踪的冗余信息。

1. 基于消费者标识过滤 #

可通过消费者的clientId、sessionId、consumerId等属性设置过滤规则，以Log4j2配置为例：

<Logger name="org.apache.activemq.artemis.jms" level="DEBUG" additivity="false">
    <AppenderRef ref="JMS_TRACE"/>
    <Filters>
        <!-- 过滤指定clientId的消费者 -->
        <StringMatchFilter stringToMatch="clientId=your-target-client-id" onMatch="ACCEPT" onMismatch="DENY"/>
        <!-- 可选：替换为consumerId过滤 -->
        <!-- <StringMatchFilter stringToMatch="consumerId=your-target-consumer-id" onMatch="ACCEPT" onMismatch="DENY"/> -->
    </Filters>
</Logger>

2. 基于队列/地址过滤 #

如果需要锁定特定队列，可通过队列名称设置过滤规则：

<Logger name="org.apache.activemq.artemis.jms" level="DEBUG" additivity="false">
    <AppenderRef ref="JMS_TRACE"/>
    <Filters>
        <StringMatchFilter stringToMatch="queue=your-target-queue-name" onMatch="ACCEPT" onMismatch="DENY"/>
    </Filters>
</Logger>

二、自定义消息拦截器获取完整信息 #

若需要精准提取JSON格式消息内容、会话信息、代理接收时间、代理投递时间，仅靠日志过滤无法满足需求，可通过实现ActiveMQ Artemis的拦截器接口，在消息流转节点插入自定义逻辑：

1. 实现拦截器类 #

创建类实现org.apache.activemq.artemis.core.server.Interceptor，分别处理消息接收（客户端到代理）和投递（代理到消费者）场景：

import org.apache.activemq.artemis.core.server.Interceptor;
import org.apache.activemq.artemis.core.protocol.core.Packet;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionSendMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionDeliverMessage;
import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;

import java.util.Date;

public class TargetMessageInterceptor implements Interceptor {

    // 可配置化的目标消费者clientId
    private String targetClientId = "your-target-client-id";
    // 可配置化的目标队列名称
    private String targetQueue = "your-target-queue";

    @Override
    public boolean intercept(Packet packet, RemotingConnection connection) throws ActiveMQException {
        // 处理客户端发送到代理的消息
        if (packet instanceof SessionSendMessage) {
            SessionSendMessage sendPacket = (SessionSendMessage) packet;
            Message msg = sendPacket.getMessage();
            SimpleString queueAddress = msg.getAddress();
            String clientId = connection.getClientID();

            if (targetQueue.equals(queueAddress.toString()) && targetClientId.equals(clientId)) {
                logMessageDetails("RECEIVED", msg, connection, new Date());
            }
        }
        // 处理代理投递到消费者的消息
        else if (packet instanceof SessionDeliverMessage) {
            SessionDeliverMessage deliverPacket = (SessionDeliverMessage) packet;
            Message msg = deliverPacket.getMessage();
            String clientId = connection.getClientID();

            if (targetClientId.equals(clientId)) {
                logMessageDetails("DELIVERED", msg, connection, new Date());
            }
        }
        return true;
    }

    private void logMessageDetails(String action, Message msg, RemotingConnection connection, Date timestamp) {
        // 提取JSON格式消息内容
        String jsonContent = new String(msg.getBodyBuffer().readBytes());
        // 提取会话ID
        String sessionId = connection.getSessionID().toString();

        // 输出定制化日志
        System.out.printf("[%s] 时间: %s, 会话ID: %s, 客户端ID: %s, 消息ID: %s, JSON内容: %s%n",
                action, timestamp, sessionId, connection.getClientID(), msg.getMessageID(), jsonContent);
    }
}

2. 注册拦截器 #

在broker.xml中配置拦截器，让代理加载并生效：

<core xmlns="urn:activemq:core">
    <!-- 其他核心配置 -->
    <interceptors>
        <interceptor-class>com.your.package.TargetMessageInterceptor</interceptor-class>
    </interceptors>
</core>

三、注意事项 #

• 日志过滤方式实现快速，但默认日志可能不打印消息内容，需确保日志级别设为DEBUG，必要时可修改Artemis的日志模板。
• 自定义拦截器可高度定制，但需注意性能影响，避免在拦截逻辑中执行耗时操作。
• 基于clientId过滤时，需确保客户端连接时明确设置该属性，否则无法生效。

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