实时 vs 近实时
eventd v3 的指标计算分为两层：
| 层级 | 延迟 | 计算方式 | 精度 |
|---|---|---|---|
| 实时（面板层） | < 1s | Redis + SSE 推送 | 粗（最新值） |
| 近实时（Flink 层） | < 1min | 窗口聚合后写入 Redis | 精确（窗口统计） |
实时层直接从 Redis 读取最新值，Flink 层定期刷新。面板看到的是实时层的最新快照，同时显示近实时层的历史趋势。
QPS 计数
Flink 内置的 numRecordsInPerSecond/numRecordsOutPerSecond 是算子级指标，eventd v3 还维护业务级 QPS：
public class QpsCounter extends KeyedProcessFunction<String, Event, QpsMetric> {
private transient Counter eventCounter;
private transient ValueState<Integer> secondCount;
private transient ValueState<Long> lastSecond;
@Override
public void open(OpenContext parameters) throws Exception {
eventCounter = getRuntimeContext()
.getMetricGroup()
.counter("events");
secondCount = getRuntimeContext().getState(
new ValueStateDescriptor<>("secondCount", Types.INT));
lastSecond = getRuntimeContext().getState(
new ValueStateDescriptor<>("lastSecond", Types.LONG));
}
@Override
public void processElement(Event event, Context ctx, Collector<QpsMetric> out)
throws Exception {
eventCounter.inc();
long currentSecond = ctx.timerService().currentProcessingTime() / 1000;
Long last = lastSecond.value();
Integer count = secondCount.value();
if (last == null || currentSecond > last) {
// 进入新秒，输出上一秒的 QPS
if (last != null) {
out.collect(new QpsMetric(ctx.getCurrentKey(), last, count));
}
lastSecond.update(currentSecond);
secondCount.update(1);
} else {
secondCount.update(count + 1);
}
// 注册 1 秒后触发的 Timer（确保最后一秒也被输出）
ctx.timerService().registerProcessingTimeTimer((currentSecond + 1) * 1000);
}
@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<QpsMetric> out)
throws Exception {
Integer count = secondCount.value();
Long last = lastSecond.value();
if (last != null && timestamp / 1000 > last) {
out.collect(new QpsMetric(ctx.getCurrentKey(), last, count));
secondCount.clear();
lastSecond.clear();
}
}
}
public record QpsMetric(String key, long second, int count) {}
这里用 KeyedProcessFunction 而不是窗口，原因是 QPS 要的是按处理时钟连续计数并立即输出：窗口按边界对齐，空闲时不会产出，且窗口输出自带延迟（要等 Watermark 或处理时间越过边界）。Flink 窗口本身可以小到毫秒级，「窗口最小 1 秒」是误解。
延迟分位数
eventd v3 需要实时计算延迟的 P50/P95/P99。Flink 的 Histogram 指标适合观测，但取值要经 MetricsReporter 导出；如果要在作业内部直接产出分位数，eventd v3 用固定容量的样本列表 + 定时排序实现：
public class LatencyPercentileFunction
extends KeyedProcessFunction<String, Event, LatencyPercentile> {
private static final int MAX_SAMPLES = 1000;
private transient ListState<Long> latencies;
@Override
public void open(OpenContext parameters) throws Exception {
latencies = getRuntimeContext().getListState(
new ListStateDescriptor<>("latencies", Types.LONG));
}
@Override
public void processElement(Event event, Context ctx, Collector<LatencyPercentile> out)
throws Exception {
long latency = System.currentTimeMillis() - event.eventTime();
latencies.add(latency);
// 保持样本数量在限制内
List<Long> values = new ArrayList<>();
latencies.get().forEach(values::add);
while (values.size() > MAX_SAMPLES) {
values.remove(0); // 移除最老的样本
}
latencies.update(values);
// 对齐到下一个整分钟再注册，避免每条记录都注册一个不同时间戳的 Timer
long now = ctx.timerService().currentProcessingTime();
ctx.timerService().registerProcessingTimeTimer(
(now / 60_000 + 1) * 60_000);
}
@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<LatencyPercentile> out)
throws Exception {
List<Long> values = new ArrayList<>();
latencies.get().forEach(values::add);
Collections.sort(values);
if (!values.isEmpty()) {
out.collect(new LatencyPercentile(
ctx.getCurrentKey(),
percentile(values, 50),
percentile(values, 95),
percentile(values, 99),
values.size()
));
}
}
private long percentile(List<Long> sorted, int p) {
int index = (int) Math.ceil(sorted.size() * p / 100.0) - 1;
return sorted.get(Math.max(0, index));
}
}
public record LatencyPercentile(String key, long p50, long p95, long p99, int sampleSize) {}
两个代价要提前知道：每条记录都把整个 ListState 取出、删除最老样本再整体写回，事件量大时这条路径本身就是瓶颈；MAX_SAMPLES = 1000 时 P99 只由最后 10 个样本决定，样本越少尾部越不稳。要求严格分位数就换成 DescriptiveStatisticsHistogram 指标导出，或使用专门的 sketch 库。
| 分位 | 含义 | 告警阈值 |
|---|---|---|
| P50 | 50% 的延迟低于此值（中位数） | < 200ms |
| P95 | 95% 的延迟低于此值 | < 1s |
| P99 | 99% 的延迟低于此值 | < 3s |
错误率实时统计
public class ErrorRateFunction
extends KeyedProcessFunction<String, Event, ErrorRateMetric> {
// 按 key 各自计数：MetricGroup 的 Counter 是整个子任务共享且只增不减的，
// 用它算错误率会得到「本子任务所有 key 的累计值」，不是当前 key 的本分钟值。
private transient ValueState<Long> successCount;
private transient ValueState<Long> errorCount;
private transient ValueState<Long> lastReportTime;
@Override
public void open(OpenContext parameters) throws Exception {
successCount = getRuntimeContext().getState(
new ValueStateDescriptor<>("success", Types.LONG));
errorCount = getRuntimeContext().getState(
new ValueStateDescriptor<>("errors", Types.LONG));
lastReportTime = getRuntimeContext().getState(
new ValueStateDescriptor<>("lastReport", Types.LONG));
}
@Override
public void processElement(Event event, Context ctx, Collector<ErrorRateMetric> out)
throws Exception {
long success = successCount.value() == null ? 0 : successCount.value();
long errors = errorCount.value() == null ? 0 : errorCount.value();
if (event.status() == 0) {
success++;
} else {
errors++;
}
successCount.update(success);
errorCount.update(errors);
long now = ctx.timerService().currentProcessingTime();
Long last = lastReportTime.value();
if (last == null || now - last >= 60_000) {
long total = success + errors;
double rate = total > 0 ? (double) errors * 100 / total : 0;
out.collect(new ErrorRateMetric(ctx.getCurrentKey(), rate, total));
// 输出后清零，得到的是「这一分钟」而不是「从启动至今」
successCount.clear();
errorCount.clear();
lastReportTime.update(now);
}
}
}
public record ErrorRateMetric(String key, double errorRatePercent, long totalCount) {}
这种「到达时判断」的实现在流量停下来后不会再输出，最后一次统计会停在半截。要保证停流后也有结果，就改成在 processElement 注册处理时间 Timer，在 onTimer 里输出并清零——与上面 QPS 的 Timer 思路一致。
指标聚合到 Redis
eventd v3 的指标写入流程：
Flink 算子计算实时指标
→ RedisMetricSink 写入 Redis
→ key: eventd:metrics:<metric>:<windowStart>:<dimensionValue>
→ TTL: 300s（5 分钟过期）
Redis 数据结构选择：
| 指标类型 | Redis 类型 | 理由 |
|---|---|---|
| 最新值 | String | 简单，O(1) 读取 |
| 历史趋势 | Sorted Set (ZADD key score timestamp) | 按时间排序 |
| 多维统计 | Hash | field = dimension, value = count |
// 写入 Sorted Set（时间序列）
public void writeTimeSeries(String metricType, String dimension, double value, long timestamp) {
String key = "eventd:metrics:ts:" + metricType + ":" + dimension;
sync.zadd(key, timestamp, timestamp + ":" + value);
sync.expire(key, 86400); // 24 小时过期
}
面板查询
实时面板的查询逻辑：
// 读取最近 5 分钟的指标
List<Map<String, Object>> getRecentMetrics(String type, String dimension) {
String key = "eventd:metrics:ts:" + type + ":" + dimension;
long now = Instant.now().toEpochMilli();
// Lettuce 的 zrangebyscore 返回 List<V>，不是 Set
List<String> values = sync.zrangebyscore(key, now - 300000, now);
List<Map<String, Object>> result = new ArrayList<>();
for (String v : values) {
String[] parts = v.split(":");
Map<String, Object> point = new HashMap<>();
point.put("timestamp", Long.parseLong(parts[0]));
point.put("value", Double.parseDouble(parts[1]));
result.add(point);
}
return result;
}
小结
| 指标 | 实现 | 精度 |
|---|---|---|
| QPS | KeyedProcessFunction + Processing Time Timer | 秒级 |
| 延迟分位数 | ListState + 排序 | 近似（1000 样本） |
| 错误率 | ValueState 计数 + 到达时输出 | 分钟级 |
| 实时查询 | Redis String | 最新值 |
| 历史趋势 | Redis Sorted Set | 时序 |
下一节 10.2 实时告警规则引擎 ：阈值判断、滑动窗口告警、告警抑制。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。