设计要点
SensorSimulatorPublisher 模拟一个传感器：元素到达间隔服从指数分布，平均到达率为 lambda 次/秒。它不是「定时器 + 无限缓冲」，而是严格按下游请求量发射。
要实现正确，至少需要处理五件事：
| 问题 | 处理方式 |
|---|---|
| 需求累加溢出 | 饱和加法，超过 Long.MAX_VALUE 时钳位 |
并发 request 导致多个发射循环 | wip 计数器保证同一时刻只有一个 drain |
| 取消后立即停止 | cancel() 置位并中断正在睡眠的工作线程 |
request(n <= 0) | 转成 onError(IllegalArgumentException)，并终止 |
| 信号串行 | 所有 onNext 都从同一个 drain 线程发出 |
完整实现
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
public final class SensorSimulatorPublisher implements Publisher<Double> {
private final double lambda;
private final Executor executor;
public SensorSimulatorPublisher(double lambda, Executor executor) {
if (!(lambda > 0) || Double.isInfinite(lambda)) {
throw new IllegalArgumentException("lambda must be finite and > 0");
}
this.lambda = lambda;
this.executor = executor;
}
@Override
public void subscribe(Subscriber<? super Double> subscriber) {
if (subscriber == null) throw new NullPointerException("subscriber");
subscriber.onSubscribe(new SensorSubscription(subscriber));
}
private final class SensorSubscription implements Subscription {
private final Subscriber<? super Double> downstream;
private final AtomicLong requested = new AtomicLong();
private final AtomicInteger wip = new AtomicInteger();
private final AtomicBoolean cancelled = new AtomicBoolean();
private final AtomicBoolean terminated = new AtomicBoolean();
// 非正 request 的错误不在调用线程直接发，交给 drain 串行交付
private final AtomicReference<Throwable> failure = new AtomicReference<>();
private volatile Thread worker;
SensorSubscription(Subscriber<? super Double> downstream) {
this.downstream = downstream;
}
@Override
public void request(long n) {
// §3.6：取消/终止之后 request 是 no-op，包括 n <= 0
if (cancelled.get() || terminated.get()) return;
if (n <= 0) {
// 不在调用线程直接 onError：会与 drain 线程的 onNext 并发
failure.compareAndSet(null,
new IllegalArgumentException("non-positive request: " + n));
scheduleDrain();
return;
}
addDemand(n);
scheduleDrain();
}
private void scheduleDrain() {
if (wip.getAndIncrement() == 0) {
try {
executor.execute(this::drain);
} catch (Throwable t) {
error(t);
}
}
}
@Override
public void cancel() {
if (cancelled.compareAndSet(false, true)) {
Thread t = worker;
if (t != null) t.interrupt();
}
}
private void drain() {
worker = Thread.currentThread();
int missed = 1;
try {
for (;;) {
Throwable f = failure.get();
if (f != null) {
error(f);
return;
}
long demand = requested.get();
long emitted = 0;
while (emitted < demand && !cancelled.get() && !terminated.get()) {
f = failure.get();
if (f != null) {
error(f);
return;
}
try {
Thread.sleep(nextIntervalMillis());
} catch (InterruptedException e) {
if (cancelled.get()) return;
Thread.currentThread().interrupt();
error(e);
return;
}
if (cancelled.get() || terminated.get()) return;
double value = 20 + ThreadLocalRandom.current().nextGaussian() * 5;
downstream.onNext(value);
emitted++;
}
if (emitted != 0) requested.addAndGet(-emitted);
missed = wip.addAndGet(-missed);
if (missed == 0) return;
}
} catch (Throwable t) {
error(t);
} finally {
worker = null;
}
}
private long nextIntervalMillis() {
double u = ThreadLocalRandom.current().nextDouble();
if (u == 0.0) u = Double.MIN_VALUE;
return Math.max(0L, (long) (-Math.log(u) / lambda * 1000));
}
private void addDemand(long n) {
for (;;) {
long r = requested.get();
if (r == Long.MAX_VALUE) return;
long updated = r + n;
if (updated < 0) updated = Long.MAX_VALUE;
if (requested.compareAndSet(r, updated)) return;
}
}
private void error(Throwable t) {
if (terminated.compareAndSet(false, true) && !cancelled.get()) {
downstream.onError(t);
}
}
}
}
几个容易漏掉的细节：
ThreadLocalRandom.current()必须在 drain 线程里取。把它保存成字段、跨线程使用会退化成共享的随机数生成器，既慢又可能产生争用。- 先检查
cancelled再onNext。cancel()可能发生在睡眠期间；中断醒来后必须重新检查，不能直接发元素。 cancel()要中断 worker。只置标志位的话，一次指数分布的长间隔（例如 2 秒）会让取消延迟到睡眠结束。error只发一次。terminated的 CAS 同时承担「非正请求只报一次」和「下游抛异常后不再发射」两个职责。
实测一：每次只请求 1 个
消费者在 onNext 里处理 200ms，然后请求下一个：
ExecutorService executor = Executors.newSingleThreadExecutor();
SensorSimulatorPublisher publisher = new SensorSimulatorPublisher(10, executor);
long start = System.nanoTime();
publisher.subscribe(new Subscriber<Double>() {
private Subscription subscription;
private int count;
@Override public void onSubscribe(Subscription s) {
this.subscription = s;
s.request(1);
}
@Override public void onNext(Double value) {
count++;
long elapsed = (System.nanoTime() - start) / 1_000_000;
System.out.printf("[%4d ms] #%02d value=%.2f%n", elapsed, count, value);
if (count < 10) {
try { Thread.sleep(200); } catch (InterruptedException e) {
Thread.currentThread().interrupt();
return;
}
subscription.request(1);
}
}
@Override public void onError(Throwable t) { t.printStackTrace(); }
@Override public void onComplete() { System.out.println("complete"); }
});
Thread.sleep(4_000);
executor.shutdownNow();
一次实测：
[ 275 ms] #01 value=23.96
[ 647 ms] #02 value=21.72
[ 894 ms] #03 value=14.84
[1187 ms] #04 value=27.63
[1464 ms] #05 value=17.16
[1849 ms] #06 value=17.88
[2222 ms] #07 value=23.52
[2495 ms] #08 value=24.69
[2986 ms] #09 value=22.80
[3220 ms] #10 value=15.68
平均每个元素约 322ms。它由两部分组成：
- 指数分布的平均间隔
1/lambda = 100ms； - 消费者处理时间 200ms。
两者串行，所以均值接近 300ms。这也解释了为什么「生产者 10/s、消费者 5/s」时，实际速率既达不到 10/s，也达不到 5/s：每个元素都要先等到达、再等处理。
实测二：一次请求 5 个
消费者不在 onNext 里睡眠，而是一次请求 5 个：
#1 at 72 ms
#2 at 238 ms
#3 at 488 ms
#4 at 713 ms
#5 at 873 ms
总耗时约 873ms，接近 5 × 100ms。这里的瓶颈是传感器的到达间隔，不是下游处理。请求量只是「允许生产多少」，不会让生产者超速。
实测三：非正请求与取消
request(0) 必须发 onError(IllegalArgumentException)：
error IllegalArgumentException: non-positive request: 0
取消后不能再收到元素：
got 1
got 2
done
第二次 onNext 里调用 cancel()，之后没有 onComplete，也没有更多值。无限流本来就不会完成；取消是唯一的正常停止方式。
与 Reactor 的等价写法
Reactor 的 Flux.create 可以表达同一个需求驱动模型，但关键是使用 sink.onRequest，而不是 scheduleAtFixedRate：
double lambda = 10;
Flux<Double> sensor = Flux.create(sink -> {
ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
sink.onRequest(n -> {
for (long i = 0; i < n && !sink.isCancelled(); i++) {
double u = ThreadLocalRandom.current().nextDouble();
long delay = Math.max(0L,
(long) (-Math.log(u == 0 ? Double.MIN_VALUE : u) / lambda * 1000));
scheduler.schedule(
() -> sink.next(20 + ThreadLocalRandom.current().nextGaussian() * 5),
delay, TimeUnit.MILLISECONDS);
}
});
sink.onDispose(() -> scheduler.shutdownNow());
}, FluxSink.OverflowStrategy.ERROR);
同样的消费者（每次请求 1 个、处理 200ms）实测：
[ 326 ms] #01 value=15.07
[ 727 ms] #02 value=18.68
[1019 ms] #03 value=24.29
[1256 ms] #04 value=23.45
[1585 ms] #05 value=18.55
[1845 ms] #06 value=22.97
[2117 ms] #07 value=22.12
[2401 ms] #08 value=13.83
[2720 ms] #09 value=20.81
[2990 ms] #10 value=37.52
这段代码有两个边界必须说明：
- 它只适合有界请求。如果下游一次请求
Long.MAX_VALUE，onRequest里的循环会尝试调度几乎无限个任务。生产实现应该把「请求量」转成「速率预算」，而不是为每个元素创建一个定时任务。 scheduleAtFixedRate不是等价写法。固定周期发射完全忽略下游需求，再配OverflowStrategy.BUFFER就变成无界缓冲；它看起来更短，但背压语义已经丢了。
如果不需要 Poisson 到达，只需要固定频率，直接用 Flux.interval 更合适；如果需要在固定频率上叠加背压策略，明确选择 onBackpressureDrop 或 onBackpressureLatest，不要用无界 buffer 掩盖问题。
小结
| 关注点 | 结论 |
|---|---|
| 到达模型 | 指数分布间隔，-ln(U)/lambda |
| 需求 | 饱和累加，按 request 限量发射 |
| 并发 | wip 保证单线程 drain |
| 取消 | 置位 + 中断 worker，睡眠后重新检查 |
| 非正请求 | onError(IllegalArgumentException)，且只报一次 |
| 耗时 | 每次只请求 1 个时约为「平均到达间隔 + 处理时间」；允许预取时接近两者中较大的那个 |
阅读导航
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。