Java日志体系深度重构指南：从基础规范到生产级调优

// 使用Jackson优化序列化
ObjectMapper mapper = new ObjectMapper();
String jsonLog = mapper.writeValueAsString(new PaymentEvent(
    UUID.randomUUID().toString(),
    Instant.now(),
    "PAYMENT_SUCCESS",
    Map.of("orderId", "ORD123", "amount", 999.99, "currency", "USD")
));

syntax = "proto3";

message LogEntry {
  string trace_id = 1;
  google.protobuf.Timestamp timestamp = 2;
  enum LogLevel {
    DEBUG = 0;
    INFO = 1;
    WARN = 2;
    ERROR = 3;
  }
  LogLevel level = 3;
  string message = 4;
  map<string, string> context = 5;
}

# Kubernetes Operator配置
apiVersion: opentelemetry.io/v1alpha1
kind: OpenTelemetryCollector
spec:
  config:
    receivers:
      otlp:
        protocols:
          grpc:
          http
    processors:
      batch:
        timeout: 10s
    exporters:
      jaeger:
        endpoint: jaeger-collector:14250
      logging:
    service:
      pipelines:
        traces:
          receivers: [otlp]
          processors: [batch]
          exporters: [jaeger, logging]

// 使用InheritableThreadLocal传递上下文
public class RequestContext {
    private static final InheritableThreadLocal<String> traceId = new InheritableThreadLocal<>();

    public static void setTraceId(String id) {
        traceId.set(id);
    }

    public static String getTraceId() {
        return traceId.get();
    }
}

// 异步任务包装器
public <T> CompletableFuture<T> withContext(Supplier<CompletableFuture<T>> task) {
    String currentTraceId = RequestContext.getTraceId();
    return task.get().thenApply(result -> {
        RequestContext.setTraceId(currentTraceId);
        return result;
    });
}

<AsyncLogger name="com.example.order" level="info" additivity="false">
    <AppenderRef ref="KafkaAppender"/>
</AsyncLogger>

<AsyncAppender name="KafkaAppender" blocking="false">
    <KafkaProducerConfig bootstrapServers="kafka:9092"/>
    <Topic name="application-logs"/>
    <Layout class="ch.qos.logback.classic.PatternLayout">
        <Pattern>%d{ISO8601} [%thread] %-5level %logger{36} - %msg%n</Pattern>
    </Layout>
</AsyncAppender>

# Hadoop压缩配置
hadoop jar log-processor.jar \
  -Dmapreduce.output.fileoutputformat.compress=true \
  -Dmapreduce.output.fileoutputformat.compress.codec=org.apache.hadoop.io.compress.SnappyCodec \
  -input /logs/2024-05 \
  -output /logs/processed/2024-05

public class SensitiveDataFilter {
    private static final Pattern ID_CARD_PATTERN = Pattern.compile("(\\d{6})\\d{8}(\\d{4})");
    
    public static String mask(String input) {
        return ID_CARD_PATTERN.matcher(input)
            .replaceAll("$1********$2");
    }
}

// 日志输出示例
logger.info("用户信息: {}", SensitiveDataFilter.mask("320123199901011234"));
// 输出：用户信息: 320123********1234

CREATE TABLE audit_logs (
    log_id UUID PRIMARY KEY,
    event_time TIMESTAMPTZ NOT NULL,
    user_id VARCHAR(36) NOT NULL,
    action TEXT NOT NULL,
    details JSONB
) PARTITION BY RANGE (event_time);

CREATE TABLE audit_logs_2024 PARTITION OF audit_logs
    FOR VALUES FROM ('2024-01-01') TO ('2025-01-01');

-- 自动归档策略
ALTER TABLE audit_logs 
ATTACH PARTITION audit_logs_2025 
FOR VALUES FROM ('2025-01-01') TO ('2026-01-01');

from pyspark.ml.feature import VectorAssembler
from pyspark.ml.classification import IsolationForest

assembler = VectorAssembler(
    inputCols=["log_length", "error_count", "warning_ratio"],
    outputCol="features"
)

model = Pipeline(
    stages=[assembler, StandardScaler(), IsolationForest()]
).fit(logs_df)

predictions = model.transform(logs_df)
predictions.filter(predictions.prediction == -1).show()

from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense

model = Sequential()
model.add(LSTM(50, input_shape=(n_steps, 1)))
model.add(Dense(1))
model.compile(optimizer='adam', loss='mse')

history = model.fit(X_train, y_train, epochs=200, batch_size=32)