《Java 消息中间件·可靠消息与消费治理》9.3 最终一致性的度量

[produce] sent=2000 elapsed=418ms (单次运行，不作基准)
[consume] matched=2000 elapsed=17151ms (单次运行，不作基准)

[latency] sample=2000 expected=2000
[latency] min=4454ms p50=10800ms p95=16568ms p99=17076ms max=17190ms avg=10794.9ms
[latency] over_100ms=2000 (100.0%)

eventId|loanId|amountCents|producedAt|slow
run-1793253100785-L-10-evt|run-1793253100785-L-10|250|1793253101922|1

consumer.subscribe(List.of(TOPIC));
while (assigned.getCount() > 0 && System.currentTimeMillis() - joinStart < 20_000) {
    consumer.poll(Duration.ofMillis(200));
    if (!consumer.assignment().isEmpty()) {
        assigned.countDown();
    }
}
while (System.currentTimeMillis() - start < 180_000) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(200));
    for (ConsumerRecord<String, String> r : records) {
        String[] f = r.value().split("\\|", -1);
        if (f.length != 5 || !f[1].startsWith(runId + "-L-")) {
            continue;                      // 过滤历史运行的数据
        }
        long producedAt = Long.parseLong(f[3]);
        if ("1".equals(f[4])) {
            Thread.sleep(10L + (seq % 3) * 15L);   // 注入的慢路径
        }
        long consumedAt = System.currentTimeMillis();
        long latency = consumedAt - producedAt;
        latencies.add(latency);
        insertDownstream(f, producedAt, consumedAt, latency);
    }
}

static long pct(List<Long> sorted, int p) {
    int idx = (int) Math.ceil(p / 100.0 * sorted.size()) - 1;
    return sorted.get(Math.max(0, Math.min(idx, sorted.size() - 1)));
}

$ export JAVA_HOME=~/.jdks/jdk-25.0.5+8/Contents/Home
$ javac --release 25 -cp kafka-clients-4.3.0.jar -d out src/ConsistencyLab.java
$ ./run.sh ConsistencyLab run 1000 0     # 基线，不注入慢路径
$ ./run.sh ConsistencyLab run 1000 10    # 每 10 条一条慢路径

[topic] exists v2c9.consistency
[run] id=run-1793253100785 records=1000 slow_every=0 topic=v2c9.consistency
[consume] assigned after 3182ms (含消费组首次加入)
[produce] sent=1000 elapsed=5288ms (单次运行，不作基准)
[consume] matched=1000 elapsed=10574ms (单次运行，不作基准)

[latency] sample=1000 expected=1000
[latency] min=2ms p50=6ms p95=21ms p99=43ms max=80ms avg=8.5ms
[latency] over_100ms=0 (0.0%)
[recon] upstream rows=1000 sum=250150
[recon] downstream rows=1000 distinct=1000 sum=250150
[recon] missing=0 []
[recon] extra=0 []
[recon] duplicated=0 []
[recon] amount_diff=0 => PASS

[run] id=run-1793253115611 records=1000 slow_every=10 topic=v2c9.consistency
[consume] assigned after 3126ms (含消费组首次加入)
[produce] sent=1000 elapsed=5351ms (单次运行，不作基准)
[consume] matched=1000 elapsed=10597ms (单次运行，不作基准)

[latency] sample=1000 expected=1000
[latency] min=2ms p50=14ms p95=46ms p99=77ms max=132ms avg=19.2ms
[latency] over_100ms=7 (0.7%)
[recon] upstream rows=1000 sum=250150
[recon] downstream rows=1000 distinct=1000 sum=250150
[recon] amount_diff=0 => PASS

-- 缺失：上游有、下游没有
SELECT u.loan_id
  FROM upstream u
  LEFT JOIN (SELECT loan_id FROM downstream WHERE run_id = ? GROUP BY loan_id) d
    ON u.loan_id = d.loan_id
 WHERE u.run_id = ? AND d.loan_id IS NULL;

-- 重复：下游同一键出现多次
SELECT loan_id, COUNT(*) c
  FROM downstream
 WHERE run_id = ?
 GROUP BY loan_id
HAVING COUNT(*) > 1;

[recon] upstream rows=1000 sum=250150
[recon] downstream rows=1000 distinct=1000 sum=250150
[recon] missing=0 []
[recon] extra=0 []
[recon] duplicated=0 []
[recon] amount_diff=0 => PASS

[tamper] id=run-tamper-1793253132132 upstream=1000 sent=999
[tamper] matched=999

[recon] upstream rows=1000 sum=250150
[recon] downstream rows=999 distinct=998 sum=249550
[recon] missing=2 [run-tamper-1793253132132-L-1000, run-tamper-1793253132132-L-999]
[recon] extra=0 []
[recon] duplicated=1 [run-tamper-1793253132132-L-1x2]
[recon] amount_diff=-600 => FAIL