3. 结构化日志基础：Log4j/SLF4J与JSON格式日志的配置与输出

在Kafka堆积诊断场景中，非结构化的文本日志（如纯字符串拼接）难以被日志采集系统（Filebeat、Logstash、Fluentd）高效解析，也无法直接与OpenAPI性能指标进行字段级关联。结构化日志的核心思想是将每一条日志记录转化为键值对集合，并以JSON格式输出，使得机器可以快速提取字段、过滤查询、聚合分析。

3.1 为什么需要结构化日志

传统日志与结构化日志在可观测性维度上的差异十分显著：

3.2 Log4j2 JSON布局配置

Log4j2 从 2.x 版本开始内置了 JsonLayout，无需额外依赖。以下是一个生产级配置示例，输出包含Kafka消费者关键上下文的结构化日志：

<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="WARN">
  <Appenders>
    <Console name="ConsoleJSON" target="SYSTEM_OUT">
      <JsonLayout complete="false" compact="true" eventEol="true">
        <KeyValuePair key="service" value="kafka-consumer"/>
        <KeyValuePair key="environment" value="${env:ENV:-dev}"/>
      </JsonLayout>
    </Console>
    <RollingFile name="FileJSON" fileName="logs/kafka-consumer.json"
                 filePattern="logs/kafka-consumer-%d{yyyy-MM-dd}-%i.json.gz">
      <JsonLayout complete="false" compact="true" eventEol="true">
        <KeyValuePair key="service" value="kafka-consumer"/>
        <KeyValuePair key="node" value="${sys:hostname}"/>
      </JsonLayout>
      <Policies>
        <TimeBasedTriggeringPolicy/>
        <SizeBasedTriggeringPolicy size="256 MB"/>
      </Policies>
    </RollingFile>
  </Appenders>

  <Loggers>
    <Logger name="com.example.kafka.consumer" level="INFO" additivity="false">
      <AppenderRef ref="ConsoleJSON"/>
      <AppenderRef ref="FileJSON"/>
    </Logger>
    <Root level="WARN">
      <AppenderRef ref="ConsoleJSON"/>
    </Root>
  </Loggers>
</Configuration>

3.3 Logback + logstash-logback-encoder 实现JSON输出

如果项目使用SLF4J + Logback（Spring Boot默认），推荐引入 net.logstash.logback:logstash-logback-encoder 库，它提供了比Logback原生Pattern更灵活的JSON编码器：

<!-- pom.xml 依赖 -->
<dependency>
    <groupId>net.logstash.logback</groupId>
    <artifactId>logstash-logback-encoder</artifactId>
    <version>7.4</version>
</dependency>

<!-- logback-spring.xml -->
<configuration>
  <appender name="JSON_FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
    <file>logs/kafka-diagnostic.json</file>
    <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
      <fileNamePattern>logs/kafka-diagnostic-%d{yyyy-MM-dd}.%i.json.gz</fileNamePattern>
      <maxHistory>30</maxHistory>
    </rollingPolicy>
    <encoder class="net.logstash.logback.encoder.LogstashEncoder">
      <!-- 自定义字段 -->
      <customFields>{"service":"kafka-consumer","cluster":"prod-us-east"}</customFields>
      <!-- 排除MDC中不需要的键 -->
      <excludeMdcKeyName>remoteAddress</excludeMdcKeyName>
    </encoder>
  </appender>

  <logger name="com.example.kafka" level="INFO" additivity="false">
    <appender-ref ref="JSON_FILE"/>
  </logger>
</configuration>

输出示例（单行JSON）：

{"@timestamp":"2025-03-21T10:15:30.123+08:00","@version":"1","message":"消费偏移提交成功","logger_name":"com.example.kafka.consumer.ConsumerRunner","thread_name":"consumer-1","level":"INFO","level_value":20000,"service":"kafka-consumer","cluster":"prod-us-east","consumer_group":"cg-order-payment","topic":"order-events","partition":3,"current_offset":152340,"lag":0}

3.4 结构化日志与OpenAPI性能关联的桥梁

下图展示了从Kafka消费者日志到OpenAPI性能指标关联的完整数据流：

3.5 在代码中注入诊断上下文

仅靠日志框架的配置无法自动捕获Kafka消费者组、分区、偏移量等信息。需要在业务代码中通过MDC（Mapped Diagnostic Context）手动注入：

import org.slf4j.MDC;

public class KafkaConsumerDiagnostic {

    private static final Logger log = LoggerFactory.getLogger(KafkaConsumerDiagnostic.class);

    public void processRecord(ConsumerRecord<String, String> record, String consumerGroup) {
        // 注入诊断上下文
        MDC.put("consumer_group", consumerGroup);
        MDC.put("topic", record.topic());
        MDC.put("partition", String.valueOf(record.partition()));
        MDC.put("offset", String.valueOf(record.offset()));

        try {
            long start = System.currentTimeMillis();
            // 业务处理逻辑
            handleMessage(record.value());
            long elapsed = System.currentTimeMillis() - start;

            // 输出结构化日志，MDC字段自动被JsonLayout/LogstashEncoder包含
            log.info("消息处理完成, 耗时={}ms, lag={}", elapsed, computeLag(record));
        } catch (Exception e) {
            log.error("消息处理异常, 异常类型={}", e.getClass().getSimpleName(), e);
        } finally {
            // 清理MDC，避免线程复用导致上下文污染
            MDC.clear();
        }
    }
}

3.6 日志字段规范与OpenAPI关联设计

为了实现日志与OpenAPI性能指标的精确关联，建议在日志和API指标中统一以下字段命名：