事件下的链路追踪和 HTTP 不一样
HTTP 服务的链路追踪只需要保证请求头里的 X-B3-TraceId 在上下游之间透传。事件场景复杂得多：一个事件可能跨越多个 topic，每个 topic 的消费者可能再产生新事件，形成一棵树而非一条链。
这意味着两个问题被放大了：
- TraceId 丢失：消费者消费后转发到新 topic 时忘了复制事件头；
- 采样断裂：生产者侧采样率 1%，消费者侧采样率 10%，同一链路有的 span 被采样、有的没有，最终拼不出完整树。
本节用纯 JDK 模拟最小 TraceContext，展示 traceId 在多个消费者间传递、span 分层、树形重建、以及采样一致性校验和断裂检测。
TraceContext 的最小可运行实现
// EventTraceDemo.java — 事件链路追踪：traceId 跨 topic 传递与 span 树重建
import java.util.*;
public class EventTraceDemo {
// 最小 TraceContext：traceId + 当前 spanId + parentSpanId + 采样标记
record TraceContext(String traceId, String spanId, String parentSpanId,
boolean sampled) {
TraceContext childSpan(String newSpanId) {
return new TraceContext(traceId, newSpanId, this.spanId, this.sampled);
}
TraceContext forForward(String newSpanId) {
return new TraceContext(traceId, newSpanId, this.spanId, this.sampled);
}
}
record EventMessage(String topic, String payload, TraceContext trace) {}
static class Consumer {
final String name;
final double sampleRate;
int processed = 0;
Consumer(String name, double sampleRate) {
this.name = name;
this.sampleRate = sampleRate;
}
List<EventMessage> consume(EventMessage in, String... outTopics) {
processed++;
if (!in.trace().sampled()) {
System.out.println("[" + name + "] 跳过未采样消息: " + in.trace().traceId());
return Collections.emptyList();
}
String childSpanId = name + "-" + processed;
TraceContext child = in.trace().childSpan(childSpanId);
System.out.println("[" + name + "] 处理: traceId=" + child.traceId()
+ ", spanId=" + child.spanId() + ", parent=" + child.parentSpanId());
List<EventMessage> outs = new ArrayList<>();
for (String outTopic : outTopics) {
TraceContext fwd = child.forForward(childSpanId + "-fwd-" + outTopic);
outs.add(new EventMessage(outTopic, "derived:" + in.payload(), fwd));
}
return outs;
}
}
public static void main(String[] args) {
TraceContext producerTrace = new TraceContext("T-abc123", "producer-0", null, true);
EventMessage msg = new EventMessage("orders", "order-001", producerTrace);
System.out.println("=== 链路追踪跨 topic 传递 ===\n");
System.out.println("[Producer] 发出: traceId=" + msg.trace().traceId()
+ ", sampled=" + msg.trace().sampled());
// 树形链路：orders → inventory → notification + analytics
Consumer consumerA = new Consumer("ConsumerA", 1.0);
List<EventMessage> msgsB = consumerA.consume(msg, "inventory");
List<TraceContext> allSpans = new ArrayList<>();
allSpans.add(producerTrace);
allSpans.add(new TraceContext("T-abc123", "ConsumerA-1", "producer-0", true));
for (EventMessage m : msgsB) {
if (m.topic().equals("inventory")) {
Consumer consumerB = new Consumer("ConsumerB", 1.0);
List<EventMessage> msgsC = consumerB.consume(m, "notification", "analytics");
allSpans.add(new TraceContext("T-abc123", "ConsumerB-1", "ConsumerA-1-fwd-inventory", true));
for (EventMessage mc : msgsC) {
Consumer consumerTail = new Consumer("Consumer" + mc.topic(), 1.0);
consumerTail.consume(mc);
allSpans.add(new TraceContext("T-abc123",
"Consumer" + mc.topic() + "-1",
"ConsumerB-1-fwd-" + mc.topic(), true));
}
}
}
System.out.println("\n=== Span 树重建 ===");
rebuildSpanTree(allSpans);
System.out.println("\n=== 采样一致性校验 ===");
checkSamplingConsistency(1.0, 1.0, "ConsumerA");
checkSamplingConsistency(1.0, 0.1, "ConsumerB-bad");
System.out.println("\n=== traceId 丢失场景 ===");
simulateLostTraceId();
System.out.println("\n=== 链路完整性检查 ===");
checkLinkageIntegrity(allSpans);
}
static void rebuildSpanTree(List<TraceContext> spans) {
Map<String, List<String>> children = new HashMap<>();
Set<String> allSpanIds = new HashSet<>();
String root = null;
for (TraceContext t : spans) {
allSpanIds.add(t.spanId());
if (t.parentSpanId() == null) {
root = t.spanId();
} else {
children.computeIfAbsent(t.parentSpanId(), k -> new ArrayList<>()).add(t.spanId());
}
}
System.out.println("根节点: " + root);
printTree(root, children, 0);
// 检查孤儿 span：parent 不存在的节点
for (TraceContext t : spans) {
if (t.parentSpanId() != null && !allSpanIds.contains(t.parentSpanId())) {
System.out.println("[WARN] 孤儿 span: " + t.spanId()
+ " 指向不存在的 parent=" + t.parentSpanId());
}
}
}
static void printTree(String node, Map<String, List<String>> children, int depth) {
System.out.println(" ".repeat(depth) + node);
if (children.containsKey(node)) {
for (String c : children.get(node)) printTree(c, children, depth + 1);
}
}
static void checkSamplingConsistency(double prodRate, double consRate, String name) {
if (prodRate != consRate) {
System.out.println("[WARN] " + name + " 采样率不一致: 生产="
+ prodRate + ", 消费=" + consRate + " → 链路可能断裂");
} else {
System.out.println("[OK] " + name + " 采样率一致");
}
}
static void simulateLostTraceId() {
TraceContext trace = new TraceContext("T-lost-001", "producer-1", null, true);
TraceContext wrong = new TraceContext("T-NEW-WRONG", "bad-span", null, true);
System.out.println("[BUG] 转发时生成新 traceId: " + wrong.traceId()
+ " (原 traceId=" + trace.traceId() + ")");
System.out.println("结果：两个 span 拼不成同一条链路，监控面板显示两条独立 trace");
}
static void checkLinkageIntegrity(List<TraceContext> spans) {
Set<String> traceIds = new HashSet<>();
for (TraceContext t : spans) traceIds.add(t.traceId());
System.out.println("链路中 traceId 数量: " + traceIds.size());
if (traceIds.size() == 1) {
System.out.println("[PASS] 所有 span 属于同一 trace，链路完整");
} else {
System.out.println("[FAIL] 检测到 " + traceIds.size() + " 个不同 traceId，链路已断裂");
}
}
}
运行：
javac --release 21 EventTraceDemo.java && java EventTraceDemo
输出：
=== 链路追踪跨 topic 传递 ===
[Producer] 发出: traceId=T-abc123, sampled=true
[ConsumerA] 处理: traceId=T-abc123, spanId=ConsumerA-1, parent=producer-0
[ConsumerB] 处理: traceId=T-abc123, spanId=ConsumerB-1, parent=ConsumerA-1-fwd-inventory
[Consumernotification] 处理: traceId=T-abc123, spanId=Consumernotification-1, parent=ConsumerB-1-fwd-notification
[Consumeranalytics] 处理: traceId=T-abc123, spanId=Consumeranalytics-1, parent=ConsumerB-1-fwd-analytics
=== Span 树重建 ===
根节点: producer-0
producer-0
ConsumerA-1
ConsumerB-1
Consumernotification-1
Consumeranalytics-1
=== 采样一致性校验 ===
[OK] ConsumerA 采样率一致
[WARN] ConsumerB-bad 采样率不一致: 生产=1.0, 消费=0.1 → 链路可能断裂
=== traceId 丢失场景 ===
[BUG] 转发时生成新 traceId: T-NEW-WRONG (原 traceId=T-lost-001)
结果：两个 span 拼不成同一条链路，监控面板显示两条独立 trace
=== 链路完整性检查 ===
链路中 traceId 数量: 1
[PASS] 所有 span 属于同一 trace，链路完整
traceId 在树形传播中保持不变，spanId 逐层嵌套，parentSpanId 指向上一跳。Span 树重建用 parent 指向做 BFS，同时检查孤儿节点——如果某 span 的 parent 不存在于收集到的 spans 中，意味着上游传输或采集有遗漏。
采样策略的两种做法
| 策略 | 做法 | 优点 | 风险 |
|---|---|---|---|
| 头传递采样 | 生产者决定是否采样，消费者只读取标记 | 一致性天然保证 | 生产端决策可能不够精准 |
| 消费者本地采样 | 每个消费者自己做采样决策 | 可根据自身负载调整 | 采样率不一致导致链路断裂 |
事件场景推荐头传递采样。生产者在发第一条消息时掷骰子决定是否采样，然后把 sampled=true/false 写进事件头。所有消费者只读标记，不做二次决策。
如果某个消费者需要比生产者更高的采样率（比如审计消费者需要全量），应在另一个独立的 consumer group 里做，不影响主链路的追踪一致性。
链路断裂的三种典型原因
| 断裂类型 | 表现 | 检测方法 | 修复 |
|---|---|---|---|
| traceId 丢失 | 同一条业务链路显示为两条独立 trace | checkLinkageIntegrity 发现多 traceId | 消费者在转发时复制 traceId |
| 采样断裂 | trace 树缺失某些分支 | 采样率校验告警 | 统一使用头传递采样 |
| 孤儿 span | parentSpanId 指向不存在的节点 | rebuildSpanTree 检查孤儿 | 检查上游消费者是否正确上报 span |
三种断裂里，采样断裂最难排查。因为监控面板看起来正常——只是某些分支不见了——你以为是链路简单，实际是消费者自己决定不采样。头传递采样可以根治这个问题。
事件追踪 vs HTTP 追踪的关键差异
| 差异点 | HTTP 场景 | 事件场景 |
|---|---|---|
| 链路形状 | 链（请求→响应） | 树（一个事件触发多个下游事件） |
| 追踪头位置 | HTTP header | 事件体或事件头字段 |
| 采样一致性 | 同一次请求内天然一致 | 容易因消费者本地采样而断裂 |
| traceId 丢失风险 | 代理转发时可能丢 | 消费者转发到新 topic 时容易忘 |
| 延迟度量 | 单次请求耗时 | 端到端跨度大，需分段累计 |
事件场景下，追踪头必须放在消息体的一个固定字段里（如 trace_context），不能用 HTTP header 的思维。消费者在收到消息后，解析出 trace 信息，在转发到新 topic 时原样复制。
为什么 traceId 不能靠日志拼接
有些团队的做法是在每个消费者里打印日志，然后用时间戳把日志串成一条"伪链路"。这种方法在事件驱动架构下是不可靠的。
事件场景下，一个事件可能触发十个下游消费者，它们彼此不知道对方存在。如果只靠日志时间戳拼接，分叉的链路会碎成十条独立的日志流。traceId 把这些独立流串成一棵有根的树——根是事件生产，分支是每个消费者的处理。
没有 traceId，排查跨服务的事件问题就只剩 grep。
本节小结
- 事件链路追踪的形状是树不是链，traceId 必须在每个转发节点原样复制。
- 采样策略推荐头传递模式：生产者一次决策，消费者只读不判。
- 采样率不一致是链路断裂的最常见原因，应在启动时校验。
- Span 重建需检查孤儿节点：parent 不存在的 span 意味着上游采集遗漏。
- 链路完整性检查（单 traceId 验证）应在 CI 或集成测试里例行运行。
- 消费者在转发时生成新 traceId 是最隐蔽的 bug，应在代码审查时重点检查。
- traceId 应使用 UUID v4 或雪花算法生成，不要用自增整数——自治节点间自增会冲突。
- Span 记录应包含时间戳字段，用于延迟分析，不能只记录层级关系。
阅读导航：上一节 9.3 事件存储 ｜ 下一节 10.2 端到端延迟度量
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。