从同步到异步：什么变了
上一节的 SyncSinglePublisher 在调用线程同步发射。真实数据源通常异步：网络响应、数据库驱动回调、消息队列拉取、定时器触发。
异步化带来三个新问题：
request(n)在消费者线程调用，onNext在生产线程调用——两者可能并发，额度记账必须是原子的。cancel()可能从任意线程到达——必须原子地停止生产，且不能阻塞调用者。- 信号必须串行（§1.3）——即使内部有多个生产线程，到达同一个 Subscriber 的
onNext不能重叠。
AsyncRangePublisher 的设计
AsyncRangePublisher 发射整数区间 [from, to)，发射逻辑提交到 Executor。它演示四个机制：
request额度累加（饱和，防溢出）- 额度扣减与发射的原子配合
- 单线程 Executor 保证信号串行
done标志保证onComplete/onError只发一次
完整实现
package com.plume.eventd.core;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
/**
* 异步范围 Publisher：在 Executor 上按 request 额度发射 [from, to)。
*
* <p>调用方必须传入单线程 Executor（或串行化的 Executor），
* 否则 §1.3 的信号串行无法保证。
*/
public final class AsyncRangePublisher implements Publisher<Integer> {
private final int from;
private final int to;
private final Executor executor;
public AsyncRangePublisher(int from, int to, Executor executor) {
if (from > to) throw new IllegalArgumentException("from > to");
this.from = from;
this.to = to;
this.executor = executor;
}
@Override
public void subscribe(Subscriber<? super Integer> subscriber) {
if (subscriber == null) throw new NullPointerException("subscriber");
subscriber.onSubscribe(new RangeSubscription(subscriber, from, to, executor));
}
private static final class RangeSubscription implements Subscription {
private final Subscriber<? super Integer> subscriber;
private final int to;
private final Executor executor;
private int current;
private final AtomicLong requested = new AtomicLong();
private final AtomicBoolean cancelled = new AtomicBoolean();
private final AtomicBoolean done = new AtomicBoolean();
private final AtomicBoolean emitting = new AtomicBoolean();
private volatile Throwable error;
RangeSubscription(Subscriber<? super Integer> subscriber,
int from, int to, Executor executor) {
this.subscriber = subscriber;
this.current = from;
this.to = to;
this.executor = executor;
}
@Override
public void request(long n) {
if (cancelled.get() || done.get()) {
return; // §3.6：cancel 后 request 是 no-op
}
if (n <= 0) {
// 不能在这里直接调 onError：request 可能来自另一个线程，
// 会与发射线程的 onNext 并发。记下错误，交给 emitLoop 串行交付
error = new IllegalArgumentException(
"§3.9: non-positive request: " + n);
tryEmit();
return;
}
if (!addDemand(n)) {
return;
}
tryEmit();
}
@Override
public void cancel() {
// §3.5：快速返回，不同步做重计算；§3.7：重复 cancel 是 no-op
cancelled.set(true);
}
/** 饱和加法：任意次 request 累加都不溢出（§3.17）。 */
private boolean addDemand(long n) {
for (;;) {
long r = requested.get();
if (r == Long.MAX_VALUE) return true;
long updated = r + n;
if (updated < 0) updated = Long.MAX_VALUE;
if (requested.compareAndSet(r, updated)) return true;
}
}
private void tryEmit() {
// 同一时刻只允许一个发射任务；否则并发 request 会重复调度
if (emitting.compareAndSet(false, true)) {
executor.execute(this::emitLoop);
}
}
private void emitLoop() {
try {
for (;;) {
if (cancelled.get() || done.get()) return;
Throwable terminal = error;
if (terminal != null) {
signalError(terminal); // 与 onNext 在同一线程串行交付
return;
}
long req = requested.get();
if (req == 0) return; // 额度用完，等下一次 request
long available = (long) to - current;
if (available <= 0) {
complete(); // 元素耗尽
return;
}
int emitCount = (int) Math.min(req, available);
// 先扣额度再发射：即使中途 cancel，也只少发、不会超发
if (!requested.compareAndSet(req, req - emitCount)) {
continue; // 有并发 request，用新额度重试
}
for (int i = 0; i < emitCount; i++) {
if (cancelled.get()) return;
terminal = error;
if (terminal != null) {
signalError(terminal);
return;
}
subscriber.onNext(current++);
}
}
} catch (Throwable t) {
signalError(t); // 下游抛异常：转成 onError
} finally {
emitting.set(false);
// 退出瞬间可能有新 request 或错误到达，补一次调度
if ((requested.get() > 0 || error != null)
&& !cancelled.get() && !done.get()) {
tryEmit();
}
}
}
private void complete() {
if (done.compareAndSet(false, true) && !cancelled.get()) {
subscriber.onComplete(); // §1.7：终止信号只发一次
}
}
private void signalError(Throwable t) {
if (done.compareAndSet(false, true)) {
subscriber.onError(t);
}
}
}
}
四个关键设计决策
1. 额度累加必须饱和
long updated = r + n;
if (updated < 0) updated = Long.MAX_VALUE;
如果直接 requested.addAndGet(n)，累计需求超过 Long.MAX_VALUE 后会溢出成负数。之后 req == 0 判断失效、Math.min 得到负的发射数量，发布者可能永久挂起或直接崩溃。§3.17 允许 Publisher 把 Long.MAX_VALUE 视为无界，所以溢出时饱和到上限是正确的。
2. 先扣额度，再发射
if (!requested.compareAndSet(req, req - emitCount)) continue;
for (int i = 0; i < emitCount; i++) subscriber.onNext(current++);
反过来「先发再扣」会在并发 request 下重复计算额度：两个线程都读到 req=10，各自发 10 个，合计 20，违反 §1.1。先扣再发保证「已扣的额度 ≥ 已发的元素」，最坏情况是中途 cancel 导致少发——§1.2 明确允许少发后终止。
CAS 失败时用 continue 重新读取额度，而不是放弃。并发 request 增加了额度，新的循环会用更大的 req 继续。
3. emitting 只负责调度互斥
if (emitting.compareAndSet(false, true)) {
executor.execute(this::emitLoop);
}
request 可能被连续调用多次：第一次启动发射任务，后面几次只加额度，不应再提交任务。emitting 就是这把锁：任务运行期间为 true，finally 中置回 false，并检查是否还有未消费的额度。
注意它不保证信号串行。信号串行来自 Executor：只有单线程（或所有任务都串行执行的）Executor 才能保证两个 emitLoop 不重叠。构造器接受任意 Executor 是妥协——如果调用方传线程池，emitting 仍能让同一时刻只有一个任务在发信号，但 current++ 和 requested 的可见性依赖 Executor 的提交语义；更稳妥的做法是在类型上就要求 ScheduledExecutorService 单线程包装。生产代码里 Reactor 用 Sinks 内部的 WIP 计数器解决同一问题，不依赖调用方。
4. done 防止终止信号重入
没有 done 时会出现一个隐蔽死循环：元素耗尽后 onComplete 被调用，但 requested 仍大于 0；finally 看到「还有额度且未取消」，再次调度 emitLoop；新任务发现 available <= 0，又一次 onComplete……只要 Subscriber 请求过超过元素总数的额度，这个循环就会持续提交任务、持续重复终止信号，违反 §1.7，还会把 Executor 队列打满。
done.compareAndSet(false, true) 让 complete() 和 signalError() 互斥且只执行一次，finally 也检查 done，循环终止。
运行验证
基本用法
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import com.plume.eventd.core.AsyncRangePublisher;
public class AsyncRangeTest {
public static void main(String[] args) throws Exception {
ExecutorService executor = Executors.newSingleThreadExecutor();
new AsyncRangePublisher(1, 6, executor).subscribe(new Subscriber<>() {
private Subscription subscription;
private int count;
@Override
public void onSubscribe(Subscription s) {
this.subscription = s;
System.out.println("onSubscribe");
s.request(2);
}
@Override
public void onNext(Integer i) {
System.out.println("onNext: " + i);
if (++count == 2) {
subscription.request(3);
}
}
@Override public void onError(Throwable t) { t.printStackTrace(); }
@Override public void onComplete() { System.out.println("onComplete"); }
});
Thread.sleep(200); // 演示用；生产代码不要用 sleep 等异步结果
executor.shutdown();
}
}
输出：
onSubscribe
onNext: 1
onNext: 2
onNext: 3
onNext: 4
onNext: 5
onComplete
前两个来自 request(2)，后三个来自 onNext(2) 里补的 request(3)。所有 onNext 都跑在 Executor 线程上。
取消后停止
@Override
public void onNext(Integer i) {
System.out.println("onNext: " + i);
if (i == 3) subscription.cancel();
}
输出 onNext: 1 到 onNext: 3，之后没有 onComplete。cancel 置位后，发射循环在下一个元素前检查并退出。
空范围
new AsyncRangePublisher(5, 5, executor)：第一次 request 后 available = 0，直接 onComplete。输出只有 onComplete。
三个并发细节
细节一：onComplete 可能早于 request 返回
同步 Executor（如 Runnable::run）下，request(10) 会在返回前完成全部发射和 onComplete。异步 Executor 下，request 返回时可能一个元素都还没发。Subscriber 不能假设「request 返回即已收到」——这是 2.9/2.10 要求「无论有无 request 都要准备接收终止信号」的现实原因。
细节二：cancel 之后仍可能收到 onNext
cancel() 只是置位，发射循环正在执行的 onNext 不会被打断；如果循环已经进入 for 且通过了 cancelled 检查，当前这个元素仍会送达。这符合 §2.8：Subscriber 必须准备好接收 cancel 之后、已经请求过的迟到元素。
细节三：下游抛异常必须转成 onError
catch (Throwable t) {
signalError(t);
}
§2.13 要求 Subscriber 的四个方法正常返回；如果它违反了（例如 onNext 里抛 RuntimeException），Publisher 不能把异常吞掉，也不能让发射线程带着异常退出。正确做法是终止订阅并（如果尚未终止）调用 onError。这里 signalError 用 done 保证不会在 onComplete 之后再发 onError。
小结
| 机制 | 实现 | 对应规则 |
|---|---|---|
| 额度累加 | 饱和 CAS | §3.8、§3.17 |
| 先扣后发 | compareAndSet | §1.1 |
| 调度互斥 | emitting | 避免重复任务 |
| 终止唯一 | done | §1.7 |
| 取消 | cancelled | §3.12 |
下一节转向消费端：实现 LoggingSubscriber，看 Subscriber 一侧的规则如何落到代码。
阅读导航
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。