Reactor 核心：背压、调度与取消

Reactor 的"响应式"不是回调糖：它是一份订阅契约——下游用 request(n) 声明自己的消化能力，上游按这个量生产，cancel 随时终止。这份契约决定了背压怎么传、线程在哪切、资源怎么放。本文固定 reactor-core 3.8.8 / reactive-streams 1.0.4（由 Boot4.0.9 BOM 解析，即 reactor-bom 2025.0.8 线）与 JDK21；全部观察来自reactor-core实验的真实运行，验收记录见 docs/audits/java-depth-u26-verification.md。WebFlux/WebClient/R2DBC 的应用链路归U27，本文不展开；STOMP消息层的发送侧缓冲与本文的订阅契约分层，见WebSocket/STOMP原文；ThreadLocal与线程池的上下文传播见异步执行原文。

订阅与请求量：一对契约方法

发布/订阅模型里，Publisher.subscribe(Subscriber) 之后两者之间只有一个中介：Subscription。它只有两个方法，各管一件事：request(n) 声明"我还能收n个"，cancel() 宣布"我不再收了"。reactive-streams 1.0.4 规范把规则编号到每一条（如 request 的n必须大于0、cancel幂等、不可以在信号里同步调用另一侧方法）；Reactor 的参考实现是 BaseSubscriber。原码：reactor-core v3.8.8 BaseSubscriber#request/requestUnbounded/cancel 节选。

原码java

	@Override
	public final void request(long n) {
		if (Operators.validate(n)) {
			Subscription s = this.subscription;
			if (s != null) {
				s.request(n);
			}
		}
	}

	/**
	 * {@link #request(long) Request} an unbounded amount.
	 */
	public final void requestUnbounded() {
		request(Long.MAX_VALUE);
	}

	@Override
	public final void cancel() {
		if (Operators.terminate(S, this)) {
			try {
				hookOnCancel();
			}
			catch (Throwable throwable) {
				hookOnError(Operators.onOperatorError(subscription, throwable, currentContext()));
			}
			finally {
				safeHookFinally(SignalType.CANCEL);
			}
		}
	}

三个读法：request 入口的 Operators.validate(n) 拦截非正数（n<=0 时向发 onError(IllegalArgumentException)，这是规范 rule 3.9 的实现）；requestUnbounded() 就是 request(Long.MAX_VALUE)——"无限量"不是魔法值，是把这个数一路加到饱和；cancel 用 Operators.terminate 原子置换Subscription引用保证幂等，只成功一次才触发 hookOnCancel。null检查同样在契约里：onNext(null)/onError(null) 直接 Objects.requireNonNull 抛出（规范 rule 2.13），不是信号。

请求量语义是理解后面一切的地基：unbounded（Long.MAX_VALUE）意味着"你要多快发多快"——大多数冷源会直接走快路径一口气发完；有界请求意味着上游发满n个就必须停下等下一次request。实验用 BaseSubscriber 手动 request(2) 订阅 range(1,10)：恰好2个onNext、0次complete——range 的慢路径发满2个就返回，complete永远不来。take(2) 则是被库包好的"收满即cancel"：实验实测源头只产生2个（emitted=2）、向上游cancel=1。

实验代码java

package study.reactor;

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import org.junit.jupiter.api.Test;
import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;

/** U26组2：request(n) 契约——有界请求控制发射节奏；take(n) 提前取消；flatMap 预取向上游放大请求。 */
class RequestContractTest {

    @Test
    void boundedRequestControlsRangeEmission() {
        List<Integer> received = new CopyOnWriteArrayList<>();
        AtomicInteger completed = new AtomicInteger();
        BaseSubscriber<Integer> subscriber = new BaseSubscriber<>() {
            @Override
            protected void hookOnSubscribe(Subscription s) {
                request(2); // 只请求2，之后不补
            }

            @Override
            protected void hookOnNext(Integer value) {
                received.add(value);
            }

            @Override
            protected void hookOnComplete() {
                completed.incrementAndGet();
            }
        };
        Flux.range(1, 10).subscribe(subscriber);
        assertEquals(List.of(1, 2), received); // range 只交付请求量
        assertEquals(0, completed.get()); // 请求量未覆盖完就永远不 complete
        subscriber.dispose();
        System.out.println("U26OBSERVE request-contract request2 received=" + received.size() + " completed=" + completed.get());
    }

    @Test
    void takeCancelsUpstreamEarly() {
        AtomicInteger upstreamCancelled = new AtomicInteger();
        AtomicInteger emitted = new AtomicInteger();
        List<Integer> received = new CopyOnWriteArrayList<>();
        Flux.range(1, 10)
                .doOnNext(i -> emitted.incrementAndGet())
                .doOnCancel(upstreamCancelled::incrementAndGet)
                .take(2)
                .subscribe(received::add);
        assertEquals(List.of(1, 2), received);
        assertEquals(2, emitted.get()); // 源头只产生了2个
        assertEquals(1, upstreamCancelled.get()); // take(2) 收满即向上游 cancel
        System.out.println("U26OBSERVE request-contract take2 emitted=" + emitted.get() + " upstreamCancel=" + upstreamCancelled.get());
    }

    @Test
    void flatMapPrefetchesBeyondDownstreamDemand() {
        AtomicLong upstreamRequested = new AtomicLong();
        List<Integer> received = new CopyOnWriteArrayList<>();
        Flux.range(1, 64)
                .doOnRequest(n -> upstreamRequested.accumulateAndGet(n, Long::sum))
                .flatMap(Mono::just)
                .subscribe(new BaseSubscriber<Integer>() {
                    @Override
                    protected void hookOnSubscribe(Subscription s) {
                        request(2); // 下游只要2
                    }

                    @Override
                    protected void hookOnNext(Integer value) {
                        received.add(value);
                    }
                });
        assertTrue(upstreamRequested.get() >= 32,
                "flatMap 应向上游预取至少小缓冲量32，实测 " + upstreamRequested.get());
        System.out.println("U26OBSERVE request-contract prefetch downstreamRequest=2 upstreamRequested=" + upstreamRequested.get());
    }

    @Test
    void stepVerifierDrivesRequestExplicitly() {
        StepVerifier.create(Flux.range(1, 10), 2)
                .expectNext(1, 2)
                .thenRequest(3)
                .expectNext(3, 4, 5)
                .thenCancel()
                .verify(); // reactor-test 的按步 request 验证：初始2、补3、取消
        System.out.println("U26OBSERVE request-contract stepVerifier request2then3 thenCancel");
    }
}

冷流与热流：每个订阅者独立执行还是共享

冷流（range/just/defer/大多数操作符组合）在每次 subscribe 时独立执行一遍生产逻辑：订阅两次=生产两次。实验里 range(1,3).doOnNext(计数) 订阅两回，副作用总计数6（3×2）——没有任何共享。热流一旦开始发射与订阅者无关，订阅者只看到"订阅之后"的值。两者不是语法差异而是生产时点差异：冷流的生产被订阅动作触发，热流的生产有自己的生命周期。

Sinks 是手工构造热流的入口，三个变体的边界要分清。Sinks.many().replay().all()：实验先发1、2，早订阅者收1、2，再发3，后到订阅者补订时收到1,2,3全部历史（replay=订阅时重放缓存），之后双方同步收4。Sinks.many().multicast().directBestEffort()：订阅前emit实测返回 FAIL_ZERO_SUBSCRIBER（值丢弃），订阅者只收订阅后的 [2,3]——live语义，历史不补。publish() 返回的 ConnectableFlux 把"何时开始"从订阅移到 connect()：实验实测只订阅不connect时冷源执行0次，connect后才执行1次，且publish（非replay）的后到者拿不到已过的历史。热流的关键推论：背压对已经错过的值无能为力——replay缓存在内存里、directBestEffort直接丢，下游慢只影响"从现在起"的交付。

背压：request(n) 如何沿链传递

range 是最诚实的有界源。原码：reactor-core v3.8.8 FluxRange.RangeSubscription#slowPath 节选——request(n) 传递到源头后的完整算法。

原码java

		void slowPath(long n) {
			final Subscriber<? super Integer> a = actual;

			long f = end;
			long e = 0;
			long i = index;

			for (; ; ) {

				if (cancelled) {
					return;
				}

				while (e != n && i != f) {

					a.onNext((int) i);

					if (cancelled) {
						return;
					}

					e++;
					i++;
				}

				if (cancelled) {
					return;
				}

				if (i == f) {
					a.onComplete();
					return;
				}

				n = requested;
				if (n == e) {
					index = i;
					n = REQUESTED.addAndGet(this, -e);
					if (n == 0) {
						return;
					}
					e = 0;
				}
			}
		}

读法：内层while最多发n个（n是本次请求量）就跳出；REQUESTED.addAndGet(this, -e) 把已发的数量从累计请求里扣掉，扣完为0就return（停止发射）；扣完还有剩余（说明等待期间下游又request了），把剩余作为新的n继续。两个并发细节：requested 是volatile的（request可能来自别的线程），每次onNext后检查 cancelled（cancel在发射间隙生效——这就是取消能停住同步源的原因）。request(Long.MAX_VALUE) 则不走slowPath而走fastPath一口气循环到底。这份实现把"请求量就是流量上限"落到了每一行。

中间操作符对request的改写有两类。过滤类（filter等）：上游来的值可能被滤掉，所以它向上游请求的量≥下游请求量，按比例放大。预取类（flatMap/reduce等）：实验实测下游只要2个、flatMap向range累计请求了256——flatMap默认预取量就是 Queues.SMALL_BUFFER_SIZE（256，reactor.bufferSize.small 系统属性可调）；这个256不是把下游的2"改"成256，而是flatMap自己作为"下游"向源头要的量，然后以自己的节奏向真下游交付。结论：链上每一对相邻operator之间的request量都可能不同，契约只在每一跳上成立。

溢出策略：Buffer、Drop、Latest、Error 的丢弃语义

onBackpressure系列操作符的前提先说清：它们向上游请求无限量、向下按下游的真实节奏交付——溢出处理的是"上游已推到我这、而下游demand=0"的那部分值。四种策略各回答"多余的值去哪"：

Drop：多余的值直接丢弃，可选 onDrop 回调观察。实验 interval(2ms) + 订阅者request(20)不补：received=20、dropped=232——请求量之外全丢，流不终止。
Latest：只保留"最新的一个"多余的值，前面的被后面的覆盖。实验用 publishOn(single,1) 把交付线程与定时器解耦后，慢订阅者（每个值忙等200ms）收到的序列是 [0,101,204,307]——忙等期间的中间值全被最新值覆盖，交付的是"最新的"，跳号是它的签名。
Buffer：全部进无界队列，不丢——代价是内存。实验慢消费下收到值连续无跳号（第i+1个恰为第i个加1），与Latest的跳号形成直接对照。onBackpressureBuffer(maxSize) 可加界，界满走 BufferOverflowStrategy：ERROR（抛异常）/DROP_LATEST（丢新）/DROP_OLDEST（丢老）。
Error：第一个多余值就转 onError 终止整条流。实验：request(5)后第6个tick触发 reactor.core.Exceptions$OverflowException，流终止。它的实现就是一行：onBackpressureDrop(t -> { throw Exceptions.failWithOverflow(); })。3.8.8 的Flux上没有 onBackpressureFatal 操作符——"致命"边界就是Error与有界Buffer的ERROR策略，不存在第三种。

一个容易被忽略的事实：这四种策略只在"下游不按时request"时才有戏份。实验对照组证明：同样的interval+慢消费，不用任何onBackpressure操作符、订阅者每消化1个补request(1)时，interval的tick被压到消费速率（pending恒为0）——因为interval的tick在定时器线程串行执行且onNext同步交付，慢消费天然拖住上游。堆积只在"有缓冲/队列把生产和消费解耦"时才发生（见反例章）。

实验代码java

package study.reactor;

import java.time.Duration;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

import org.junit.jupiter.api.Test;
import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;

/**
 * U26组3：背压策略对照——interval 定时源 + 有界请求慢消费下，Drop/Latest/Buffer/Error 的收到/丢弃行为。
 * interval 的 tick 由 parallel 定时器线程驱动；onBackpressureXxx 向上游请求无限量、向下按下游节奏交付，多余值按策略处理。
 */
class BackpressureStrategiesTest {

    private static final Duration TICK = Duration.ofMillis(2);

    @Test
    void dropDiscardsEverythingBeyondRequest() throws Exception {
        AtomicLong received = new AtomicLong();
        AtomicLong dropped = new AtomicLong();
        BaseSubscriber<Long> sub = new BaseSubscriber<>() {
            @Override
            protected void hookOnSubscribe(Subscription s) {
                request(20); // 请求量之外全部丢弃，不补请求
            }

            @Override
            protected void hookOnNext(Long value) {
                received.incrementAndGet();
            }
        };
        Flux.interval(TICK).onBackpressureDrop(d -> dropped.incrementAndGet()).subscribe(sub);
        Thread.sleep(500);
        sub.dispose();
        assertEquals(20, received.get()); // 恰好收到请求量
        assertTrue(dropped.get() >= 100, "实测丢弃 " + dropped.get());
        System.out.println("U26OBSERVE backpressure-drop received=" + received.get() + " dropped=" + dropped.get());
    }

    @Test
    void latestKeepsMostRecentDuringSlowConsumer() throws Exception {
        List<Long> received = new CopyOnWriteArrayList<>();
        BaseSubscriber<Long> sub = new BaseSubscriber<>() {
            @Override
            protected void hookOnSubscribe(Subscription s) {
                request(1);
            }

            @Override
            protected void hookOnNext(Long value) {
                received.add(value);
                try { // 慢消费：忙等200ms后补1个请求
                    Thread.sleep(200);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                request(1);
            }
        };
        // publishOn(prefetch=1) 解耦交付线程：忙等期间定时器不受拖累继续推进，Latest 层只保留最新
        Flux.interval(TICK)
                .onBackpressureLatest()
                .publishOn(Schedulers.single(), 1)
                .subscribe(sub);
        Thread.sleep(700);
        sub.dispose();
        assertTrue(received.size() >= 2, "received=" + received);
        assertEquals(0L, received.get(0));
        long second = received.get(1);
        assertTrue(second >= 50, "第二个收到的应是忙等期间的最新值（跳号），实测 " + second);
        System.out.println("U26OBSERVE backpressure-latest received=" + received);
    }

    @Test
    void bufferKeepsEverythingConsecutively() throws Exception {
        List<Long> received = new CopyOnWriteArrayList<>();
        BaseSubscriber<Long> sub = new BaseSubscriber<>() {
            @Override
            protected void hookOnSubscribe(Subscription s) {
                request(1);
            }

            @Override
            protected void hookOnNext(Long value) {
                received.add(value);
                try { // 慢消费：20ms一个
                    Thread.sleep(20);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                request(1);
            }
        };
        Flux.interval(TICK).onBackpressureBuffer().subscribe(sub);
        Thread.sleep(500);
        sub.dispose();
        assertFalse(received.isEmpty());
        for (int i = 1; i < received.size(); i++) { // 无跳号：堆积在缓冲中不丢失
            assertEquals(received.get(i - 1) + 1, received.get(i),
                    "第" + i + "个值不连续：" + received);
        }
        System.out.println("U26OBSERVE backpressure-buffer consecutiveCount=" + received.size()
                + " first=" + received.get(0) + " last=" + received.get(received.size() - 1));
    }

    @Test
    void errorSignalsOverflowWhenDemandExhausted() throws Exception {
        AtomicLong received = new AtomicLong();
        AtomicReference<Throwable> error = new AtomicReference<>();
        CountDownLatch errored = new CountDownLatch(1);
        BaseSubscriber<Long> sub = new BaseSubscriber<>() {
            @Override
            protected void hookOnSubscribe(Subscription s) {
                request(5);
            }

            @Override
            protected void hookOnNext(Long value) {
                received.incrementAndGet();
            }

            @Override
            protected void hookOnError(Throwable t) {
                error.set(t);
                errored.countDown();
            }
        };
        Flux.interval(TICK).onBackpressureError().subscribe(sub);
        assertTrue(errored.await(5, TimeUnit.SECONDS), "应在demand用尽后收到onError");
        sub.dispose();
        assertEquals(5, received.get());
        assertTrue(error.get() != null && error.get().toString().toLowerCase().contains("overflow"),
                "实际异常：" + error.get());
        System.out.println("U26OBSERVE backpressure-error received=" + received.get()
                + " errorType=" + error.get().getClass().getName());
    }
}

调度器：publishOn 与 subscribeOn 各管哪一段

两个操作符都换线程，但作用边界完全不同，混用是常见事故源。subscribeOn(s) 影响订阅信号向上传播的线程——对 range 这类同步源，就是源头onSubscribe/request/发射所在的线程。publishOn(s) 影响它下游的onNext/onError/onComplete交付线程——对它上游怎么执行没有发言权。实验用线程名实测对照（Thread.currentThread().getName() 采集）：

range.doOnNext(采集).publishOn(boundedElastic).doOnNext(采集)：前段全在 main（订阅线程），后段全在 boundedElastic-1。
range.doOnNext(采集).subscribeOn(newSingle("u26-sub"))：源头段全在 u26-sub-2；去掉subscribeOn则在 main。
组合 range.doOnNext(source段).publishOn(boundedElastic).doOnNext(mid段).subscribeOn(single)：source段=u26-sub-3（subscribeOn管源头）、mid段=boundedElastic-1（publishOn管它以下的交付）——两个操作符各改各的段，互不越界。

三个内置调度器的用途：parallel()（N=CPU核数的专用worker，为非阻塞CPU密集工作准备）；single()（单线程串行，适合需要严格顺序的交付）；boundedElastic()（唯一为阻塞调用准备的弹性池，默认上限10×CPU线程数与100000排队任务，由 reactor.schedulers.defaultBoundedElasticSize/defaultBoundedElasticQueueSize 系统属性可调；JDK21+ 上还可通过 reactor.schedulers.defaultBoundedElasticOnVirtualThreads=true 切到虚拟线程承载——默认仍为平台线程池）。Flux.parallel(2).runOn(Schedulers.parallel()) 是并行分片操作符，把流拆成2条轨道在parallel上执行再 sequential() 合并——实验实测所有轨道线程都在 parallel-N。调度器是全局缓存的单例（Schedulers.parallel() 每次返回同一实例），newSingle 这类工厂创建的是独占实例，用完要 dispose()——实验的自定义调度器都在finally里释放。

阻塞隔离：parallel 上为什么不能 block

Reactor 用线程标记而不是静态分析来隔离阻塞：parallel/single 调度器的工作线程实现 NonBlocking 标记接口，block() 入口检测当前线程，命中即抛异常。原码：reactor-core v3.8.8 BlockingSingleSubscriber#blockingGet 节选。

原码java

	final @Nullable T blockingGet() {
		if (Schedulers.isInNonBlockingThread()) {
			throw new IllegalStateException("block()/blockFirst()/blockLast() are blocking, which is not supported in thread " + Thread.currentThread().getName());
		}
		if (getCount() != 0) {
			try {
				await();
			}
			catch (InterruptedException ex) {
				dispose();
				Thread.currentThread().interrupt();
				throw Exceptions.propagate(ex);
			}
		}

		Throwable e = error;
		if (e != null) {
			RuntimeException re = Exceptions.propagate(e);
			//this is ok, as re is always a new non-singleton instance
			re.addSuppressed(new Exception("#block terminated with an error"));
			throw re;
		}
		return value;
	}

实验实测：publishOn(parallel) 之后的 Mono.just("x").map(v->v).block() 抛 IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread parallel-7——消息带线程名，与源码逐字一致。但有一个必须如实记录的边界：Mono.just("x").block()（不带任何中间operator）不抛——因为 MonoJust 覆写了 block() 直接 return value，根本不进入 BlockingSingleSubscriber 的检测（MonoCallable 同理，fromCallable(...).block() 直接调用callable）。这不是"parallel上block有时安全"，而是"同步标量没有等待、没有阻塞"的实现细节；任何中间operator（map/hide）都让链脱离MonoJust、恢复检测。结论的正确表述：parallel/single线程上任何需要等待的block都会被抛出拦截；boundedElastic与自建线程不实现NonBlocking，可以承载阻塞。

承载方式实验：Mono.fromCallable(() -> { Thread.sleep(150); return 线程名; }).subscribeOn(boundedElastic).block()——callable真实sleep在 boundedElastic-1 上（耗时160ms可证），block本身在main线程调用是允许的（main不实现NonBlocking）。错误姿势的反面教材：在parallel的onNext里直接 Thread.sleep 不会有人拦你——阻塞隔离只保护block族API，不保护任意阻塞调用，在parallel里写阻塞代码是绕过检测的静默事故（会拖死worker），靠评审与 Schedulers.isInNonBlockingThread() 自查。

错误是信号：终止语义与 try-catch 的差异

onError 不是异常处理，是终止信号：一旦发出，流进入终止态，之后的值不再产生、complete不再来。实验：just(1,2,3).map(i -> { if (i==2) throw …; return i; }) 订阅结果是 [next:1, error:mid-failure]——第3个值没有产生、没有complete。与try-catch的心智差异：try-catch捕获后可以继续执行后续语句；onError之后整条流终局，想"继续"必须显式换轨——onErrorResume(e -> 备选流)。实验实测 onErrorResume 的信号序列：[next:1, resume:mid-failure, next:99, complete]——resume之后切换到备选流并正常complete，原流后续值（3）丢弃。map里抛出的异常会被operator边界转成onError（这就是"操作符边界即catch边界"）；只有fatal异常（如OOM）穿透。onErrorContinue 是少数"跳过坏值继续"的逃生舱，但它只对支持它的operator生效且语义争议大，生产慎用。

重试：Backoff 的真实时间间隔

retry() 无参版本立刻重订阅，对"间歇性故障"是自毁式风暴；真实世界用 retryWhen(Retry.backoff(maxAttempts, minBackoff))。实验用会失败两次、第三次成功的源配 Retry.backoff(5, 200ms)：3次尝试、两次重试间隔实测 276ms与315ms（总耗时606ms）——不是200ms，因为backoff默认 jitter=0.5：实际间隔 = minBackoff × (1 + random×0.5)，落在[200, 300)ms。要确定性的间隔用 Retry.fixedDelay（min=max、无jitter）。重试的消耗要计量：每次重订阅对冷流意味着重新执行整段生产（下游看到的副作用会重复），不可幂等的操作（扣款、发消息）必须把重试预算与幂等键一起设计。

耗尽的语义实验：Retry.backoff(2, 30ms) 配总失败源——实测3次尝试（1次初始+2次重试，maxAttempts 计的是重试次数），最后抛 RetryExhaustedException(Retries exhausted: 2/2)，cause链里是原始的 IllegalStateException(boom-3)。耗尽后上游错误被包装——想拿原始异常要walk getCause链，实验的异常链打印 RetryExhaustedException ← IllegalStateException 即为此设计的断言。

实验代码java

package study.reactor;

import java.time.Duration;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.util.retry.Retry;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;

/** U26组5：错误/重试——错误信号终止语义；retryWhen(Backoff) 的真实时间间隔；耗尽后 onError；与try-catch的差异。 */
class RetryBackoffTest {

    @Test
    void errorSignalTerminatesStream() {
        List<String> signals = new CopyOnWriteArrayList<>();
        AtomicInteger mapped = new AtomicInteger();
        Flux.just(1, 2, 3)
                .map(i -> {
                    mapped.incrementAndGet();
                    if (i == 2) {
                        throw new IllegalStateException("mid-failure");
                    }
                    return i;
                })
                .subscribe(v -> signals.add("next:" + v),
                        e -> signals.add("error:" + e.getMessage()),
                        () -> signals.add("complete"));
        assertEquals(List.of("next:1", "error:mid-failure"), signals); // 出错即终止：3不再产生、无complete
        assertEquals(2, mapped.get());
        System.out.println("U26OBSERVE retry errorTerminal signals=" + signals);
    }

    @Test
    void onErrorResumeBranches() {
        List<String> signals = new CopyOnWriteArrayList<>();
        Flux.just(1, 2, 3)
                .map(i -> {
                    if (i == 2) {
                        throw new IllegalStateException("mid-failure");
                    }
                    return i;
                })
                .onErrorResume(e -> {
                    signals.add("resume:" + e.getMessage());
                    return Flux.just(99);
                })
                .subscribe(v -> signals.add("next:" + v),
                        e -> signals.add("error:" + e.getMessage()),
                        () -> signals.add("complete"));
        assertEquals(List.of("next:1", "resume:mid-failure", "next:99", "complete"), signals);
        System.out.println("U26OBSERVE retry onErrorResume signals=" + signals);
    }

    @Test
    void backoffWaitsRealTimeBetweenAttempts() {
        AtomicInteger attempts = new AtomicInteger();
        List<Long> attemptNanos = new CopyOnWriteArrayList<>();
        Mono<String> flaky = Mono.defer(() -> {
            attempts.incrementAndGet();
            attemptNanos.add(System.nanoTime());
            if (attempts.get() < 3) {
                return Mono.error(new IllegalStateException("fail-" + attempts.get()));
            }
            return Mono.just("ok");
        });
        long start = System.nanoTime();
        String result = flaky.retryWhen(Retry.backoff(5, Duration.ofMillis(200))).block();
        long totalMs = (System.nanoTime() - start) / 1_000_000;
        assertEquals("ok", result);
        assertEquals(3, attempts.get());
        long gap1Ms = (attemptNanos.get(1) - attemptNanos.get(0)) / 1_000_000;
        long gap2Ms = (attemptNanos.get(2) - attemptNanos.get(1)) / 1_000_000;
        // Retry.backoff 默认 jitter=0.5：实际间隔在 [200, 300) ms
        assertTrue(gap1Ms >= 190, "第一次重试间隔实测 " + gap1Ms + "ms");
        assertTrue(gap2Ms >= 190, "第二次重试间隔实测 " + gap2Ms + "ms");
        assertTrue(totalMs >= 380, "总耗时实测 " + totalMs + "ms");
        System.out.println("U26OBSERVE retry backoff gapsMs=" + gap1Ms + "," + gap2Ms
                + " totalMs=" + totalMs + " attempts=" + attempts.get());
    }

    @Test
    void retryExhaustionSurfacesOnError() {
        AtomicInteger attempts = new AtomicInteger();
        Mono<String> alwaysFails = Mono.defer(() -> {
            attempts.incrementAndGet();
            return Mono.error(new IllegalStateException("boom-" + attempts.get()));
        });
        RuntimeException thrown = assertThrows(RuntimeException.class,
                () -> alwaysFails.retryWhen(Retry.backoff(2, Duration.ofMillis(30))).block());
        AtomicReference<Throwable> walk = new AtomicReference<>(thrown);
        StringBuilder chain = new StringBuilder();
        while (walk.get() != null) {
            chain.append(walk.get().getClass().getSimpleName())
                    .append("(").append(walk.get().getMessage()).append(")")
                    .append(walk.get().getCause() == null ? "" : " <- ");
            if (walk.get().getCause() == null) {
                break;
            }
            walk.set(walk.get().getCause());
        }
        assertTrue(attempts.get() >= 2, "至少初始+重试：" + attempts.get());
        assertTrue(chain.toString().contains("boom"), "异常链应含原始错误，实测 " + chain);
        System.out.println("U26OBSERVE retry exhausted attempts=" + attempts.get()
                + " chain=" + chain + " exceptionType=" + thrown.getClass().getSimpleName());
    }
}

取消：dispose、take(n) 与 timeout 的取消链

取消沿订阅链向上传播，每一跳的operator都有机会执行自己的收尾。三个触发入口：Disposable.dispose()（订阅返回值，订阅者主动取消）；take(n)（收满n个自动向上游cancel）；timeout(t)（超时先cancel上游、再onError给下游）。实验的触发矩阵（doOnCancel/doFinally，SignalType枚举为 ON_COMPLETE/ON_ERROR/CANCEL）：

场景	上游doOnCancel	所在位置doFinally
手动dispose	1	CANCEL
take(2)收满	1（take的上游被cancel）	上游=CANCEL、take下游=ON_COMPLETE
错误终止	0	ON_ERROR

注意3.8.8 的Flux/Mono上没有 doOnDispose 操作符（这是Disposable.dispose()的动作名，不是操作符）——订阅链上可观察的取消信号就是 doOnCancel 与 doFinally 的CANCEL类型；take收满时"上游被取消、下游正常完成"是这个矩阵里最反直觉的一行。

取消真的停得住上游吗？实验用无限源验证：Flux.interval(20ms) 订阅210ms后dispose，计数冻结在10，再等150ms仍是10——interval的worker被cancel，定时任务不再触发。timeout链路实验：fromCallable(300ms慢源).subscribeOn(boundedElastic).timeout(50ms).onErrorResume(TimeoutException, 备选)——50ms时定时器线程cancel订阅，doOnCancel=1、备选值交付；被取消的callable的sleep会被中断，那个InterruptedException以 onErrorDropped 日志形式出现（订阅已终止无处投递）——这不是失败，是取消后的正常噪音，读日志时不要误判。flatMap的inner取消实验：just("a","b").flatMap(key -> interval无限流).take(4)——take(4)的cancel穿过flatMap传播到每条inner interval，实测activeInners归零。

资源释放：using 在 cancel 时的清理

Flux.using(资源工厂, 流工厂, 清理函数) 是订阅生命周期的try-with-resources：清理在complete、error、cancel三种终止下都执行。实验：using(() -> "conn-1", conn -> interval(10ms), conn -> 清理) 订阅80ms后dispose——实测cleanup执行（cleanupThread=main，dispose的发起线程）。这回答"响应式资源谁关"：不挂在finally块（订阅是异步的，finally在订阅返回时就跑了），挂在订阅终止信号上。doOnDiscard 补充另一种泄漏面：被丢弃（Drop/缓冲淘汰）的值若持有资源，用 doOnDiscard(类型, 清理) 收尾——丢弃路径不经过任何onError/onComplete。

Context：与数据流反向的传播

Reactor的 Context 解决的是"订阅链上的隐式参数"（traceId、租户），方向设计是它全部反直觉的来源：Context从订阅者出发、沿订阅链向上游传播（与数据流方向相反）。contextWrite(ctx -> ctx.put(k,v)) 在这个传播路径上写入，于是：链上位于contextWrite之前声明的operator（更靠近源）读得到；之后声明的（更靠近订阅者）读不到——写入向上游方向可见、不向订阅者方向泄漏。实验双向断言：just.flatMap(deferContextual).contextWrite(put) 的flatMap里读到 value；just.contextWrite(put).flatMap(deferContextual) 的flatMap里读到 absent。读Context的正规入口是 Mono.deferContextual/Flux.deferContextual（或实现 CoreSubscriber.currentContext()），map/flatMap的lambda参数里没有"当前Context"可拿——它是订阅期数据，不是执行期全局状态。

跨operator传递实验：just.map(拼1).flatMap(拼2).flatMap(deferContextual拼context) 里最内层读到了 contextWrite 写入的 v26——中间任何operator都不截断。线程无关性实验：deferContextual(读).subscribeOn(boundedElastic).contextWrite(put) 切线程后仍读到 carried，而ThreadLocal在同样切换后是 null——Context随订阅传播、与执行线程无关，这正是它替代ThreadLocal的理由（ThreadLocal在异步边界断链，见异步执行原文的传播矩阵）。代价：不可变结构（每次put生成新Context）、只对Reactor链内可见——跨到非Reactor代码（线程池里的Runnable）要用ContextPropagation/Micrometer手动桥接。

实验代码java

package study.reactor;

import org.junit.jupiter.api.Test;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;

/** U26组7：Context——Context 与数据流方向相反、自订阅者向上游传播；contextWrite 写入对其上游链路可见、不向订阅者方向传播；跨 operator 传递；对照 ThreadLocal 不跨线程。 */
class ContextPropagationTest {

    @Test
    void contextWriteDirectionality() {
        // Context 自订阅者向上游（源方向）传播；contextWrite 在传播路径上写入：
        // 链上位于 contextWrite 之前声明（更靠近源）的 operator 读得到
        String upstreamOfContextWrite = Mono.just("a")
                .flatMap(v -> Mono.deferContextual(ctx -> Mono.just(String.valueOf(ctx.getOrEmpty("k").orElse("absent")))))
                .contextWrite(ctx -> ctx.put("k", "value"))
                .block();
        assertEquals("value", upstreamOfContextWrite);

        // 链上位于 contextWrite 之后声明（更靠近订阅者）的 operator 读不到：写入不向订阅者方向传播
        String downstreamOfContextWrite = Mono.just("a")
                .contextWrite(ctx -> ctx.put("k", "value"))
                .flatMap(v -> Mono.deferContextual(ctx -> Mono.just(String.valueOf(ctx.getOrEmpty("k").orElse("absent")))))
                .block();
        assertEquals("absent", downstreamOfContextWrite);
        System.out.println("U26OBSERVE context direction beforeWrite=" + upstreamOfContextWrite
                + " afterWrite=" + downstreamOfContextWrite);
    }

    @Test
    void contextTraversesOperators() {
        String result = Mono.just("a")
                .map(v -> v + "-1")
                .flatMap(v -> Mono.just(v + "-2"))
                .flatMap(v -> Mono.deferContextual(ctx -> Mono.just(v + ":" + ctx.getOrEmpty("k").orElse("none"))))
                .contextWrite(ctx -> ctx.put("k", "v26"))
                .block();
        assertEquals("a-1-2:v26", result); // map/flatMap 链条不妨碍 Context 向上传递
        System.out.println("U26OBSERVE context traversed=" + result);
    }

    @Test
    void contextSurvivesThreadSwitchUnlikeThreadLocal() {
        ThreadLocal<String> threadLocal = new ThreadLocal<>();
        threadLocal.set("mainValue");
        try {
            // ThreadLocal 对照：subscribeOn 切线程后读不到
            String fromOtherThread = Mono.fromCallable(threadLocal::get)
                    .subscribeOn(Schedulers.boundedElastic())
                    .block();
            assertNull(fromOtherThread, "ThreadLocal不跨线程");

            // Context：随订阅传播，与执行线程无关
            String carried = Mono.<String>deferContextual(ctx -> Mono.just(String.valueOf(ctx.getOrEmpty("k").orElse("absent"))))
                    .subscribeOn(Schedulers.boundedElastic())
                    .contextWrite(ctx -> ctx.put("k", "carried"))
                    .block();
            assertEquals("carried", carried);
            System.out.println("U26OBSERVE context threadLocalOnOtherThread=" + fromOtherThread
                    + " reactorContextOnOtherThread=" + carried);
        } finally {
            threadLocal.remove();
        }
    }
}

反例：无界缓冲与慢消费的堆积

把"请求量就是流量上限"忘掉的两种典型姿势，实验都可观察。姿势一：无界缓冲+慢消费。interval(2ms).doOnNext(计数).onBackpressureBuffer().publishOn(single,1) 配20ms消费一个的订阅者：500ms时实测 emitted=279、received=21、pending=258在缓冲里堆着（本次运行pending=236/emittedAtCancel=258），dispose后emitted冻结——堆积存在、可观察、可清理，但清理靠的是取消订阅而不是"消化完"：无界队列不会自己变短。生产等价物：把消息全部 onBackpressureBuffer 进内存再慢消费，OOM只是时间问题。姿势二：flatMap映射到无限流且不设界。inner流不结束、flatMap的并发轨道被占满，后续inner排队——必须给外层take/timeout，实验证明take(4)能把全部inner取消（activeInners=0）。

对照组是同一实验的第二组：同样快源慢消费，不用缓冲操作符、订阅者窗口恒为1——实测pending=0、emitted=19=received=19（tick被压到消费速率）。它揭示堆积的真实条件：堆积需要"生产和消费被缓冲解耦"；没有缓冲，慢消费自己就是背压（拖住上游）。这给选型一条硬规则：每引入一层缓冲（onBackpressureBuffer无界版、publishOn的prefetch队列、Sinks的unicast buffer），就问一句"这个队列满的时候谁受害、用什么策略"——答不出就别引入。

实验代码java

package study.reactor;

import java.time.Duration;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import org.junit.jupiter.api.Test;
import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;

/** U26组8：反例——无界请求/无界缓冲 + 慢消费的堆积：AtomicLong 计数证明 pending 堆积存在且取消后停止增长。 */
class UnboundedSlowConsumerTest {

    @Test
    void fastSourceWithSlowConsumerBacksUpInBuffer() throws Exception {
        AtomicLong emitted = new AtomicLong();
        AtomicLong received = new AtomicLong();
        CountDownLatch first = new CountDownLatch(1);
        BaseSubscriber<Long> sub = new BaseSubscriber<>() {
            @Override
            protected void hookOnSubscribe(Subscription s) {
                request(1);
            }

            @Override
            protected void hookOnNext(Long value) {
                if (received.incrementAndGet() == 1) {
                    first.countDown();
                }
                try { // 慢消费：20ms一个
                    Thread.sleep(20);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                request(1);
            }
        };
        // 发射计数在 onBackpressureBuffer 之前；publishOn 解耦交付线程：定时器不受慢消费拖累，缓冲持续堆积
        Flux.interval(Duration.ofMillis(2))
                .doOnNext(v -> emitted.incrementAndGet())
                .onBackpressureBuffer()
                .publishOn(Schedulers.single(), 1)
                .subscribe(sub);

        assertTrue(first.await(5, TimeUnit.SECONDS));
        Thread.sleep(500);
        long pending = emitted.get() - received.get();
        assertTrue(pending >= 50, "缓冲堆积应可观察，实测 pending=" + pending + "（emitted="
                + emitted.get() + " received=" + received.get() + "）");
        sub.dispose();
        long emittedAtCancel = emitted.get();
        Thread.sleep(150);
        assertEquals(emittedAtCancel, emitted.get(), "取消后堆积源头必须停止增长");
        System.out.println("U26OBSERVE counter-example pending=" + pending + " emittedAtCancel=" + emittedAtCancel
                + " received=" + received.get() + " frozenAfterCancel=true");
    }

    @Test
    void boundedRequestCapsInflightInsteadOfBuffering() throws Exception {
        // 对照组：同样的快源+慢消费，无缓冲操作符直接订阅。
        // interval 的 tick 在定时器线程串行执行且 onNext 同步交付：慢消费把 tick 压到消费速率（上游被拖住），不堆积
        AtomicLong emitted = new AtomicLong();
        AtomicLong received = new AtomicLong();
        CountDownLatch first = new CountDownLatch(1);
        BaseSubscriber<Long> sub = new BaseSubscriber<>() {
            @Override
            protected void hookOnSubscribe(Subscription s) {
                request(1); // 永远最多1个在途
            }

            @Override
            protected void hookOnNext(Long value) {
                if (received.incrementAndGet() == 1) {
                    first.countDown();
                }
                try {
                    Thread.sleep(20);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                request(1);
            }
        };
        Flux.interval(Duration.ofMillis(2))
                .doOnNext(v -> emitted.incrementAndGet()) // interval仍按自身节奏推进
                .subscribe(sub); // 无缓冲操作符：请求量就是流量上限
        assertTrue(first.await(5, TimeUnit.SECONDS));
        Thread.sleep(500);
        sub.dispose();
        long pending = emitted.get() - received.get();
        // interval 源不因下游停发（无 onBackpressureXxx 干预），但订阅者窗口恒为1：慢消费本身拖住 tick 交付
        assertTrue(pending <= 5, "请求窗口1时在途应近似为0，实测 pending=" + pending);
        System.out.println("U26OBSERVE counter-example boundedWindow1 pending=" + pending
                + " emitted=" + emitted.get() + " received=" + received.get());
    }
}

架构师变化条件：请求量预算与背压选型

K08的四问。单位与预算：请求量是"条"，背压预算要落到内存（队列长度×元素大小）与时延（排队等待时间=队列长度×消费间隔）——boundedElastic的100000排队任务上限与onBackpressureBuffer的maxSize都是这个预算的实例；flatMapping到无限流时，flatMap自身的并发上限（默认256）就是轨道预算。两个合理选择：慢消费者配Drop+降级日志（丢得明白，内存恒定）vs 配有界Buffer+ERROR（不丢则已、丢则报错）——选择依据是业务对"丢一条"与"晚一条"的容忍度，不是性能测试数字。成本与恢复责任：每层缓冲的清理责任在引入者——订阅取消、超时、错误耗尽三条终止路径都必须能触发清理（用using/doFinally验证）；重试的重复副作用由幂等键兜底，不靠"小心"。演进验证：Reactor大版本升级时本票实测的高风险点要复验——block检测的标量覆写边界、onBackpressure系列的操作符存在性（3.8.8无onBackpressureFatal/doOnDispose）、jitter默认值（backoff=0.5）；这些是API表面之下容易静默变化的语义，升级时以本票实验为回归基线（labs/reactor-core/run.py 一次调用全部复跑）。

与周边机制的分层：HTTP/WebSocket传输层有自己的流控窗口（TCP窗口、WS的发送缓冲，见STOMP原文的TERMINATE/DROP对照）；Reactor的request契约止步于订阅边界，不会自动穿透到socket——网络层的"背压"是另一套机制（U27的Reactor Netty线）。把"响应式=自动背压=不丢数据"三个等号连着念是本次升级要纠正的核心误解：响应式给的是背压的通道（request/cancel），用不用、怎么用，仍是设计者的决定。

闭卷验收

本页 15 道闭卷题收进了下方自测面板：先回忆，再展开参考答案与评分点，最后给三档自评。

Reactor 核心：背压、调度与取消 ​

订阅与请求量：一对契约方法 ​

冷流与热流：每个订阅者独立执行还是共享 ​

背压：request(n) 如何沿链传递 ​

溢出策略：Buffer、Drop、Latest、Error 的丢弃语义 ​

调度器：publishOn 与 subscribeOn 各管哪一段 ​

阻塞隔离：parallel 上为什么不能 block ​

错误是信号：终止语义与 try-catch 的差异 ​

重试：Backoff 的真实时间间隔 ​

取消：dispose、take(n) 与 timeout 的取消链 ​

资源释放：using 在 cancel 时的清理 ​

Context：与数据流反向的传播 ​

反例：无界缓冲与慢消费的堆积 ​

架构师变化条件：请求量预算与背压选型 ​

闭卷验收 ​

Reactor 核心：背压、调度与取消

订阅与请求量：一对契约方法

冷流与热流：每个订阅者独立执行还是共享

背压：request(n) 如何沿链传递

溢出策略：Buffer、Drop、Latest、Error 的丢弃语义

调度器：publishOn 与 subscribeOn 各管哪一段

阻塞隔离：parallel 上为什么不能 block

错误是信号：终止语义与 try-catch 的差异

重试：Backoff 的真实时间间隔

取消：dispose、take(n) 与 timeout 的取消链

资源释放：using 在 cancel 时的清理

Context：与数据流反向的传播

反例：无界缓冲与慢消费的堆积

架构师变化条件：请求量预算与背压选型

闭卷验收