这份基线能回答什么
它只回答两个问题：
- 在同一台机器、同一个 JDK 上，几个同步路径的量级差多少？
- 把信号交给另一个调度器时，排队和唤醒大概要多久？
它不能回答「Reactor 比手写快几倍」「生产环境能跑多少 TPS」。这类结论需要 JMH、固定 CPU、隔离负载、多轮分位数，并且要报告 JVM 参数和 GC。下面的代码没有引入 JMH，因为它不是本工程的构建依赖；它是一个可读、可复现的起点，不是基准套件。
环境
Apple M1 Pro，10 核，32 GB，macOS 26.3 (25D125)
Temurin 25.0.5+8
Apache Maven 3.9.12
Reactor 3.8.8（BOM 2025.0.8）
JVM flags: -Xms512m -Xmx1g -XX:+UseG1GC
eventd-core 需要先编译：
export JAVA_HOME=~/.jdks/jdk-25.0.5+8/Contents/Home
cd .local/java_reactive/eventd-v1
mvn -q -DskipTests compile
吞吐与订阅开销
import com.plume.eventd.core.SyncSinglePublisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.util.Arrays;
import java.util.concurrent.Flow;
public class Bench {
static volatile long sink;
public static void main(String[] args) {
int n = 1_000_000;
SyncSinglePublisher<Integer> handWritten = new SyncSinglePublisher<>(1);
Mono<Integer> mono = Mono.just(1);
Runnable range = () -> Flux.range(1, n).subscribe(v -> sink += v);
Runnable monoRuns = () -> {
for (int i = 0; i < n; i++) {
mono.subscribe(v -> sink += v);
}
};
Runnable handRuns = () -> {
for (int i = 0; i < n; i++) {
handWritten.subscribe(new Flow.Subscriber<Integer>() {
@Override public void onSubscribe(Flow.Subscription s) {
s.request(Long.MAX_VALUE);
}
@Override public void onNext(Integer v) { sink += v; }
@Override public void onError(Throwable t) { throw new RuntimeException(t); }
@Override public void onComplete() { }
});
}
};
for (int i = 0; i < 3; i++) {
range.run();
monoRuns.run();
handRuns.run();
}
measure("Flux.range(1M)", 5, range, n);
measure("Mono.just x1M", 5, monoRuns, n);
measure("SyncSinglePublisher x1M", 5, handRuns, n);
}
static void measure(String name, int runs, Runnable task, int ops) {
long[] ms = new long[runs];
for (int i = 0; i < runs; i++) {
long t0 = System.nanoTime();
task.run();
ms[i] = (System.nanoTime() - t0) / 1_000_000;
}
long[] sorted = ms.clone();
Arrays.sort(sorted);
System.out.printf("%-28s ms=%s median=%dms ops/s=%.0f%n",
name, Arrays.toString(ms), sorted[runs / 2],
ops / (sorted[runs / 2] / 1000.0));
}
}
编译时把 eventd-core/target/classes 和 Reactor 的 jar 放进 classpath。用 Maven 跑的话，把这段代码放到 eventd-app/src/main/java 下，依赖会自动带上。
本机实测：
Java: 25.0.5
warmup done
Flux.range(1M) ms=[6, 6, 6, 6, 6] median=6ms ops/s=166666667
Mono.just x1M ms=[24, 21, 24, 24, 26] median=24ms ops/s=41666667
SyncSinglePublisher x1M ms=[6, 6, 6, 6, 6] median=6ms ops/s=166666667
怎么读这张表
Flux.range(1M)是「一次订阅发 100 万个元素」，6ms 约等于 1.66 亿元素/秒。这个数字很高，因为消费者只做一次long累加，没有 I/O、没有跨线程、没有对象分配；它说明的是「同步路径的纯循环上限」，不是业务吞吐。Mono.just x1M是「订阅 100 万次，每次一个元素」，24ms 约等于每次订阅 24ns。这里测的是订阅开销，不是元素吞吐。SyncSinglePublisher x1M也是「订阅 100 万次」，6ms 约等于每次 6ns。它比Mono.just快，但两者做的不是同一件事：手写类没有操作符链、没有 Context、没有调度器、没有错误钩子，也没有经过 TCK 以外的场景验证。不要据此得出「手写比 Reactor 快 4 倍」的结论。- 5 轮的极差很小（6ms 全是 6ms），并不代表测量精确，而是任务太短、分辨率只有毫秒。要看更细的差异应改用纳秒并增大规模。
ops/s 这一列对后两行是按「订阅次数」算的，对第一行是按「元素个数」算的；把它们并排比较时要注意单位不同。
正确的数字之外，还要记录什么
至少记录：
| 项 | 为什么 |
|---|---|
| JDK 供应商与版本 | 不同 GC、不同 JIT 差异很大 |
| JVM flags | 堆大小直接影响分配与 GC |
| 预热轮数 | 不预热会把解释执行算进去 |
| 轮数与统计量 | 只报平均值会掩盖长尾 |
| 输入规模 | 1M 与 100M 的缓存行为不同 |
| 源码版本 | 同一段代码改动后不可直接比较 |
调度延迟
下面测的是「元素从上游进入 publishOn 队列，到下游 onNext 被调用」的时间，不是业务端到端延迟：
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;
import java.util.Arrays;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public class Latency {
public static void main(String[] args) throws Exception {
for (int i = 0; i < 3; i++) run(100_000);
long[] result = run(100_000);
Arrays.sort(result);
System.out.printf("p50=%dus p90=%dus p99=%dus p999=%dus max=%dus%n",
result[result.length / 2] / 1000,
result[(int) (result.length * 0.90)] / 1000,
result[(int) (result.length * 0.99)] / 1000,
result[(int) (result.length * 0.999)] / 1000,
result[result.length - 1] / 1000);
}
static long[] run(int n) throws Exception {
long[] latencies = new long[n];
AtomicInteger index = new AtomicInteger();
CountDownLatch done = new CountDownLatch(1);
Flux.range(0, n)
.map(i -> new long[]{System.nanoTime(), i})
.publishOn(Schedulers.parallel())
.subscribe(pair -> {
latencies[index.getAndIncrement()] = System.nanoTime() - pair[0];
if (index.get() == n) done.countDown();
});
if (!done.await(10, TimeUnit.SECONDS)) {
throw new IllegalStateException("timeout");
}
return latencies;
}
}
本机实测：
p50=5us p90=10us p99=11us p999=100us max=110us
读法：
- p50 5μs 是
parallel调度器唤醒一个 worker 的典型量级，不是 Reactor 的固定开销。 - p99 11μs 与 p50 接近，说明大多数元素没有排长队；p999 跳到 100μs，通常是线程被抢占或 GC。
- 这次运行把 10 万个元素一次性
request给publishOn，默认预取 256，队列里会积压；延迟随队列长度增长。要测「空载延迟」，应该让下游逐个请求。 - 每个元素都分配了一个
long[]，分配和 GC 本身会进入测量结果。更严谨的做法是复用时间戳数组或用 JMH 的Blackhole。
旧版几个测试为什么不能用
Flux.range(...).onBackpressureBuffer()
旧说法是「Flux.range 会按下游请求量发射，所以在它后面加 onBackpressureBuffer() 不会触发缓冲」。这句话是错的。onBackpressureBuffer 会向上游请求 Long.MAX_VALUE，把 range 变成推模型；缓冲区是否溢出，取决于下游补充 request 的时机：
- 下游在
onNext里同步request(1)，元素边进缓冲边被取走，容量 10 也可能永远不满； - 中间加
publishOn(..., 1)形成异步边界，1000 个元素会一次性灌入，容量 10 的缓冲立刻溢出。
所以旧测试的问题不是「操作符无效」，而是它既没有异步边界，也没有容量或溢出断言，测到的只是同步循环。要测缓冲开销，必须固定容量、固定生产速率、显式控制下游 request，并记录溢出次数。
delayElements(1ms).take(100) 测内存
take(100) 只消费 100 个元素，delayElements 每个元素延迟 1ms，总耗时约 100ms。Runtime.totalMemory() - freeMemory() 的差值包含尚未回收的临时对象，既不是峰值，也不是稳定占用。这段代码测到的是「100ms 内分配了多少东西」，不是「队列占多少内存」。
只有一次计时的延迟
long start = System.nanoTime();
Mono.just(i).block();
latencies[i] = System.nanoTime() - start;
单次 nanoTime 的分辨率和噪声都可能大于被测操作本身；block() 还可能涉及线程状态切换。至少要做预热、多轮、分位数，并明确测的是哪一段。
虚构的对比表
「无背压 ~3M/s、Buffer(100) ~2M/s、Buffer(10000) ~1M/s」这类数字没有对应的测量代码，换一台机器就不成立。更有价值的是给出测量方法，让读者在自己的环境里得到数字。
资源占用怎么测
不要用 totalMemory/freeMemory 猜。可用：
# GC 日志：分配速率、停顿、堆变化
java -Xms512m -Xmx1g -Xlog:gc*:file=gc.log:time,uptime,level,tags -cp ... Bench
# JFR：分配采样与热点
java -XX:StartFlightRecording=filename=run.jfr,settings=profile -cp ... Bench
jfr view allocation-by-class run.jfr
关注：
| 指标 | 来源 |
|---|---|
| 分配速率 | GC 日志的 Allocation Rate 或 JFR 采样 |
| 峰值堆 | GC 日志中的 Heap After GC 与 Heap Before GC |
| 停顿分布 | GC 日志的 pause 事件 |
| 队列深度 | 应用自己的指标，不是 JVM 指标 |
如果要在书里写具体内存数字，必须同时给出堆大小、GC、输入规模和测量时长；否则数字没有意义。
一个可复现的基线记录模板
日期：
机器 / CPU / 内存：
OS：
JDK：
JVM flags：
源码版本（git commit）：
预热轮数：
测量轮数：
输入规模：
指标：中位数 / p99 / 最大值
备注（是否有其他进程、是否开启 JFR）：
小结
| 测量对象 | 本次结果 | 说明 |
|---|---|---|
Flux.range 发 1M 元素 | 6ms（约 1.66 亿/秒） | 纯同步循环，消费者无工作 |
Mono.just 订阅 1M 次 | 24ms（约 24ns/次） | 订阅开销，不是元素吞吐 |
SyncSinglePublisher 订阅 1M 次 | 6ms（约 6ns/次） | 手写路径更薄，但能力更少 |
publishOn 调度延迟 | p50 5μs，p99 11μs | 10 万元素、默认预取、含分配噪声 |
这些数字只对本次运行、这台机器、这个 JDK 成立。性能章节的价值不在于给出一个可复制的数字，而在于给出可复制的方法和必须记录的上下文。
阅读导航
上一篇：14.1 用 Project Reactor 重写 eventd
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。