我们的性能监控仪表盘彻底失效了。问题出在一个运行了数年的大型单体应用上，它基于Spring Boot和JPA/Hibernate。APM工具展示的P95响应时间曲线一片平稳，但客户支持团队收到的关于“系统卡顿”的投诉却在稳步增加。传统的日志系统里充斥着无用的INFO信息，而DEBUG级别的日志一旦开启，磁盘IO就会瞬间被打满，整个系统陷入瘫瘓。我们需要的是手术刀，而我们手里只有一把生锈的锤子。
问题的核心在于，我们无法将宏观的性能指标（如平均响应时间）与微观的业务上下文（如特定的租户ID、用户行为、请求TraceID）关联起来。哪个租户的哪个操作导致了数据库的慢查询？在某个营销活动期间，哪些API的数据库访问模式发生了恶化？APM无法回答这些高基数维度的查询问题。我们需要记录下每一次SQL执行的详尽上下文，并能对其进行快速、任意维度的切片和聚合分析。
初步的构想是记录每一条由Hibernate生成的SQL，以及它的执行耗时、关联的TraceID、TenantID等业务信息，然后将这些结构化数据发送到一个能够处理海量数据的分析引擎中。ELK技术栈是第一个被否决的方案。在真实项目中，我们踩过太多它的坑：对于这种高基数、时间序列数据的聚合查询，Elasticsearch的性能和资源消耗并不理想。我们需要的是一个真正的OLAP猛兽。
这就是ClickHouse进入视野的原因。它为分析而生，列式存储带来的极致压缩率和查询性能，对高基数维度的聚合分析几乎没有压力。我们的目标数据模型非常清晰：一个包含时间戳、TraceID、租户ID、执行耗时、SQL模板、调用来源等字段的巨大宽表。
现在，只剩下一个关键问题：如何以最低的侵入性和最小的性能开销，从我们的JPA应用中捕获这些数据？
方案一：基于AOP的拦截
使用Spring AOP拦截所有Repository方法是一个直观的想法。
@Aspect
@Component
public class RepositoryPerformanceAspect {
// ... logger and other setup
@Around("execution(* com.example.repository.*.*(..))")
public Object profile(ProceedingJoinPoint pjp) throws Throwable {
long start = System.nanoTime();
try {
return pjp.proceed();
} finally {
long end = System.nanoTime();
// Log execution details
}
}
}这个方案很快被否决。它能测量整个方法的执行时间，但无法精确到Hibernate实际生成的SQL以及数据库的真实执行耗时。一个Repository方法内可能包含复杂的业务逻辑，甚至多次数据库交互。AOP的切面粒度太粗了。
方案二：定制化的JDBC Driver或DataSource代理
我们可以使用像p6spy这样的库，或者自己实现一个DataSource代理，来拦截所有JDBC调用。这能精确捕获到SQL和执行时间。
但它的问题在于集成和上下文传递。想把TraceID或TenantID这种业务信息从服务层一路传递到JDBC拦截器里，通常需要依赖ThreadLocal，这会让代码变得复杂且容易出错。更重要的是，我们真正关心的往往是经过规范化（去掉具体参数值）的SQL模板，而不是每一次执行的具体SQL，以便进行聚合分析。在JDBC层面做SQL规范化，无异于重新发明一个SQL解析器。
最终选择：Hibernate StatementInspector
在深入研究了Hibernate的内部机制后，我们找到了最理想的武器：org.hibernate.resource.jdbc.spi.StatementInspector。这是一个回调接口，Hibernate在每次执行JDBC Statement前后都会调用它。它不仅能拿到即将执行的SQL，还能在执行后精确地测量耗时。最关键的是，它运行在Hibernate的Session上下文中，我们可以非常自然地获取到所有业务信息，并且它拿到的就是Hibernate已经准备好的、最原始的SQL字符串。
这正是我们需要的那把手术刀。
步骤一：构建结构化的性能日志生产者
首先，我们定义一个POJO来承载所有需要记录的性能度量信息。这保证了日志结构的清晰和一致性。
// QueryMetrics.java
// 使用Lombok简化代码
import lombok.Builder;
import lombok.Value;
import java.time.Instant;
@Value
@Builder
public class QueryMetrics {
// 基础信息
Instant timestamp;
String traceId;
String tenantId;
String applicationName;
// SQL执行信息
String sql; // 这是Hibernate生成的原始SQL
long executionTimeNanos; // 执行耗时（纳秒）
boolean success;
String exceptionClass; // 如果执行失败
// 调用栈信息（用于定位代码源）
String sourceClass;
String sourceMethod;
}接下来是核心的StatementInspector实现。我们使用ThreadLocal来持有计时器，确保在inspect（执行前）和afterExecution（执行后）之间传递状态。
// PerformanceStatementInspector.java
package com.example.infra.hibernate;
import com.example.infra.logging.QueryMetrics;
import com.example.infra.logging.StructuredLogger;
import com.example.security.TenantContext; // 假设有一个获取租户信息的上下文
import org.hibernate.resource.jdbc.spi.StatementInspector;
import org.slf4j.MDC; // 用于获取TraceID
import java.time.Instant;
public class PerformanceStatementInspector implements StatementInspector {
private static final StructuredLogger METRICS_LOGGER = new StructuredLogger();
private final ThreadLocal<Long> startTime = new ThreadLocal<>();
@Override
public String inspect(String sql) {
startTime.set(System.nanoTime());
return sql; // 我们不对SQL做任何修改，只是记录
}
// Hibernate 6.x 引入了 afterExecution 方法
// 如果使用 Hibernate 5.x, 需要在 inspect 方法中返回一个代理 Statement 来实现类似功能，但更复杂
public void afterExecution(String sql, boolean success, Exception e) {
Long startNanos = startTime.get();
if (startNanos == null) {
// 如果某些非查询操作（如schema导出）触发了inspector，startNanos可能为空，直接忽略
return;
}
long durationNanos = System.nanoTime() - startNanos;
startTime.remove(); // 清理ThreadLocal，防止内存泄漏
// 异步或通过专用线程池记录日志，避免阻塞业务线程
// 为简化示例，这里直接记录
try {
QueryMetrics.QueryMetricsBuilder builder = QueryMetrics.builder()
.timestamp(Instant.now())
.traceId(MDC.get("traceId")) // 从SLF4J的MDC获取traceId
.tenantId(TenantContext.getCurrentTenantId()) // 从业务上下文中获取租户ID
.applicationName("my-monolith-app")
.sql(normalizeSql(sql)) // 对SQL进行规范化处理
.executionTimeNanos(durationNanos)
.success(success);
if (!success && e != null) {
builder.exceptionClass(e.getClass().getName());
}
// 获取调用栈信息是一个昂贵的操作，在生产环境中需要谨慎使用或通过采样控制
// StackWalker api from Java 9+ is more efficient
StackWalker.getInstance(StackWalker.Option.RETAIN_CLASS_REFERENCE)
.walk(s -> s.filter(f -> f.getClassName().startsWith("com.example.service"))
.findFirst()
.ifPresent(frame -> {
builder.sourceClass(frame.getClassName());
builder.sourceMethod(frame.getMethodName());
}));
METRICS_LOGGER.log(builder.build());
} catch (Exception loggingException) {
// 日志记录的异常绝对不能影响主流程
// 在真实项目中，这里应该有一个健壮的错误处理机制
}
}
private String normalizeSql(String sql) {
// 一个非常基础的SQL规范化实现，将参数替换为'?'
// 生产环境建议使用更成熟的SQL解析库如JSqlParser
// SELECT * FROM users WHERE id = 1 AND name = 'test' -> SELECT * FROM users WHERE id = ? AND name = ?
return sql.replaceAll("(?<=[ =,(])'[^']+'", "?") // 替换字符串字面量
.replaceAll("(?<=[ =,(])\\b\\d+\\b", "?"); // 替换数字字面量
}
}StructuredLogger是一个简单的封装，使用Jackson将POJO序列化为JSON字符串，并输出到专门的logger。
// StructuredLogger.java
package com.example.infra.logging;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class StructuredLogger {
// 使用一个特定的logger name，方便在logback中进行隔离配置
private static final Logger logger = LoggerFactory.getLogger("query-metrics");
private final ObjectMapper objectMapper;
public StructuredLogger() {
this.objectMapper = new ObjectMapper();
// 注册JavaTimeModule以正确序列化Instant
this.objectMapper.registerModule(new JavaTimeModule());
}
public void log(QueryMetrics metrics) {
try {
String jsonLog = objectMapper.writeValueAsString(metrics);
logger.info(jsonLog);
} catch (Exception e) {
// Handle serialization error
}
}
}接下来，配置Spring Boot使用我们的StatementInspector。
# application.yml
spring:
jpa:
properties:
hibernate.session_factory.statement_inspector: com.example.infra.hibernate.PerformanceStatementInspector最后，配置Logback。我们不希望这些JSON日志和普通的应用日志混在一起。我们将为query-metrics这个logger创建一个专用的Appender，它会把日志直接输出到控制台（在K8s等环境中，这通常意味着被日志收集代理捕获）。
<!-- logback-spring.xml -->
<configuration>
<include resource="org.springframework.boot.logging.logback.base.xml"/>
<!-- 这是一个只输出原始消息的encoder，因为我们的消息已经是JSON了 -->
<encoder class="ch.qos.logback.core.encoder.LayoutWrappingEncoder">
<layout class="ch.qos.logback.classic.layout.PatternLayout">
<pattern>%msg%n</pattern>
</layout>
</encoder>
<!-- Query Metrics Appender, a rolling file appender is better for production -->
<appender name="QUERY_METRICS_CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
<encoder class="ch.qos.logback.core.encoder.LayoutWrappingEncoder">
<layout class="ch.qos.logback.classic.layout.PatternLayout">
<pattern>%msg%n</pattern>
</layout>
</encoder>
</appender>
<!-- 将 "query-metrics" logger 的日志路由到专用的appender -->
<!-- additivity="false" 阻止日志向上传递给root logger，避免重复输出 -->
<logger name="query-metrics" level="INFO" additivity="false">
<appender-ref ref="QUERY_METRICS_CONSOLE"/>
</logger>
</configuration>至此，我们的应用已经变成了一个高性能的、结构化的SQL执行度量数据源。每一次JPA操作都会生成一条类似下面这样的JSON日志，干净、规整，可以直接被下游消费。
{"timestamp":"2023-10-27T10:45:12.345Z","traceId":"a1b2c3d4-e5f6-7890-a1b2-c3d4e5f67890","tenantId":"tenant-001","applicationName":"my-monolith-app","sql":"select u1_0.id,u1_0.name,u1_0.email from users u1_0 where u1_0.id=?","executionTimeNanos":1234567,"success":true,"exceptionClass":null,"sourceClass":"com.example.service.UserService","sourceMethod":"findById"}步骤二：ClickHouse中的数据建模
数据管道部分，我们生产环境中使用Vector从日志文件读取数据，推送到Kafka，再由ClickHouse的Kafka引擎消费。为了简化，这里我们直接展示ClickHouse的表结构。设计ClickHouse的表结构是性能的关键。
CREATE TABLE default.query_metrics_local ON CLUSTER my_cluster
(
`timestamp` DateTime64(3, 'UTC'),
`traceId` UUID,
`tenantId` String,
`applicationName` LowCardinality(String),
`sql` String,
`executionTimeNanos` UInt64,
`success` Bool,
`exceptionClass` LowCardinality(String),
`sourceClass` String,
`sourceMethod` String
)
ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/query_metrics_local', '{replica}')
PARTITION BY toYYYYMM(timestamp)
ORDER BY (applicationName, tenantId, toDate(timestamp))
SETTINGS index_granularity = 8192;
CREATE TABLE default.query_metrics ON CLUSTER my_cluster AS default.query_metrics_local
ENGINE = Distributed(my_cluster, default, query_metrics_local, rand());这里的几个设计决策至关重要：
-
ENGINE = ReplicatedMergeTree: 在生产环境中提供数据冗余和高可用。 -
PARTITION BY toYYYYMM(timestamp): 按月分区，这是最常用的时间维度剪枝策略。 -
ORDER BY (applicationName, tenantId, toDate(timestamp)): 排序键是ClickHouse性能的灵魂。我们将低基数的applicationName和tenantId放在前面，可以极大地加速按这些字段进行GROUP BY和WHERE的查询。 -
LowCardinality(String): 对低基数的字符串字段使用此类型，可以显著减少存储空间并提升查询性能。 -
Distributed表: 创建一个分布式表，使得我们可以在集群的任意节点上进行查询，ClickHouse会自动将查询分发到所有分片上并行执行。
步骤三：在Jupyter中开启探索性分析
当海量数据稳定流入ClickHouse后，真正的价值才开始显现。Jupyter Notebook是进行探索性数据分析的绝佳工具，它让SRE和开发人员能够以前所未有的灵活性和深度洞察系统行为。
这是我们的整个数据流架构：
graph TD
A[Spring Boot App] -->|JPA/Hibernate| B(Custom StatementInspector);
B --> C{QueryMetrics POJO};
C -->|Jackson| D(JSON Log);
D -->|Logback| E[Log File / stdout];
E -->|Vector/Fluentd| F[Kafka];
F -->|Kafka Engine| G[ClickHouse Cluster];
G -->|clickhouse-driver| H[Jupyter Notebook];
subgraph Observability Pipeline
E; F; G;
end
现在，让我们进入Jupyter，看看能做什么。
# 安装依赖
# !pip install clickhouse-driver pandas matplotlib
import os
from clickhouse_driver import Client
import pandas as pd
import matplotlib.pyplot as plt
# --- 连接到ClickHouse ---
# 在真实项目中，密码等敏感信息应通过环境变量或 secrets manager 管理
client = Client(
host='your-clickhouse-host.com',
user='default',
password=os.environ.get('CLICKHOUSE_PASSWORD'),
port=9000,
secure=False # or True if using TLS
)
print("ClickHouse connection successful:", client.execute('SELECT 1'))
# --- 分析场景1：定位P99延迟最高的SQL模板 ---
query_p99_latency = """
SELECT
sql,
count() as total_executions,
quantile(0.99)(executionTimeNanos) / 1000000.0 AS p99_latency_ms
FROM query_metrics
WHERE timestamp >= now() - INTERVAL 1 HOUR
GROUP BY sql
ORDER BY p99_latency_ms DESC
LIMIT 10
"""
df_p99 = pd.DataFrame(client.execute(query_p99_latency), columns=['sql', 'executions', 'p99_latency_ms'])
print("Top 10 Slowest SQL Templates (P99 Latency):")
print(df_p99)
# --- 分析场景2：分析特定高价值租户的数据库负载 ---
tenant_id_to_check = 'tenant-premium-001'
query_tenant_load = f"""
SELECT
toStartOfMinute(timestamp) AS minute,
count() AS queries_per_minute,
avg(executionTimeNanos) / 1000000.0 AS avg_latency_ms
FROM query_metrics
WHERE tenantId = '{tenant_id_to_check}' AND timestamp >= now() - INTERVAL 6 HOUR
GROUP BY minute
ORDER BY minute
"""
df_tenant = pd.DataFrame(client.execute(query_tenant_load), columns=['minute', 'qpm', 'avg_latency_ms'])
df_tenant.set_index('minute', inplace=True)
# 使用Matplotlib进行可视化
fig, ax1 = plt.subplots(figsize=(15, 6))
color = 'tab:red'
ax1.set_xlabel('Time')
ax1.set_ylabel('Queries Per Minute (QPM)', color=color)
ax1.plot(df_tenant.index, df_tenant['qpm'], color=color, marker='o', linestyle='-')
ax1.tick_params(axis='y', labelcolor=color)
ax2 = ax1.twinx()
color = 'tab:blue'
ax2.set_ylabel('Average Latency (ms)', color=color)
ax2.plot(df_tenant.index, df_tenant['avg_latency_ms'], color=color, marker='x', linestyle='--')
ax2.tick_params(axis='y', labelcolor=color)
fig.tight_layout()
plt.title(f'Database Load for Tenant: {tenant_id_to_check}')
plt.grid(True)
plt.show()
# --- 分析场景3：找出导致数据库错误的源头代码 ---
query_error_source = """
SELECT
sourceClass,
sourceMethod,
exceptionClass,
count() AS error_count
FROM query_metrics
WHERE success = false AND timestamp >= now() - INTERVAL 1 DAY
GROUP BY sourceClass, sourceMethod, exceptionClass
ORDER BY error_count DESC
LIMIT 10
"""
df_errors = pd.DataFrame(client.execute(query_error_source), columns=['class', 'method', 'exception', 'count'])
print("\nTop 10 Error Sources:")
print(df_errors)
这些在Jupyter中几行代码就能完成的分析，在过去需要花费数天时间进行日志筛选、grep和脚本处理，而且结果往往不尽人意。现在，我们拥有了对系统数据库层面行为的完全洞察力。当下次出现性能抖动时，我们可以在几分钟内定位到是哪个租户的哪个操作，由哪段代码触发的哪条SQL模板，在什么时间点开始变慢。
局限性与未来展望
这个方案并非没有成本。StatementInspector会对每一次SQL执行增加微小的开销，尽管在我们的测试中这个开销可以忽略不计，但在每秒执行数万次查询的极端场景下仍需评估。此外，获取调用栈信息是一个相对昂贵的操作，生产环境中可能需要引入采样机制，例如只对超过一定耗时的查询或随机的1%查询记录调用栈。
另一个需要注意的隐私和安全问题是SQL中的参数。我们当前的normalizeSql方法丢弃了所有参数值，这是最安全的做法。如果业务需要分析具体参数，例如某个entityId的查询分布，那么需要设计一套非常严格的脱敏和白名单机制，防止敏感数据泄露到日志系统中。
未来的迭代方向很明确。首先是自动化，将Jupyter中的探索性分析固化为自动化的监控告警。我们可以通过Grafana的ClickHouse数据源，创建实时的性能仪表盘，并对P99延迟、错误率等关键指标设置动态阈值告警。其次是关联分析，将这份SQL度量数据与应用日志、APM的Trace数据在ClickHouse中进行联合查询（JOIN），构建一个完全统一、贯穿应用和数据库的可观测性平台。
