WebFlux、R2DBC 与 RSocket：请求、事务和取消

当一个 HTTP 请求需要等待数据库和远程服务，调用方可以阻塞线程等结果，也可以把等待表示成 Publisher。后者能复用事件循环，但代价是资源、错误和身份都必须跟随订阅，不能靠方法返回或 ThreadLocal 推断完成。先学Reactor 订阅与 Context，再看本页的三条真实链路。

固定新增工程基线：JDK21、Boot4.0.9、Framework7.0.10、Reactor3.8.8、Reactor Netty1.3.8、RSocket Java1.1.5、r2dbc-mysql1.4.3、Security core7.0.8；完整实验解析实际依赖，MySQL 为本机8.4.7。与旧Framework6.1.12的Servlet请求课分开，不拼成一条执行链。

Servlet 与 Reactive：先认清运行边界

层	Servlet MVC	本实验 WebFlux
入口与调度	Servlet 容器、Filter、DispatcherServlet；异步请求有再次分派	Reactor Netty 适配 HttpHandler；WebFilter、DispatcherHandler 的订阅链
找端点	MVC HandlerMapping/HandlerAdapter	reactive 包下的 HandlerMapping/HandlerAdapter；同名接口不是同一个类型
输出	HttpMessageConverter；MVC 也可适配 Publisher 返回值	HttpMessageReader/Writer、Encoder/Decoder；JSON 或流式媒体格式决定输出
阻塞依赖	同步 JDBC/HTTP 占用执行线程，需连接/线程容量预算	R2DBC/WebClient 组合异步；JDBC 必须隔离到专用有界执行资源
隐式资源	Servlet 身份常由过滤器在 ThreadLocal 中恢复/清理；命令式事务绑定线程	身份与响应式事务关联 Reactor Context；跨线程不等于跨事务

WebFlux 也可以部署到受支持的 Servlet 容器，但使用其非阻塞适配入口，不因此经过 DispatcherServlet。本页实跑的是 Netty，不能冒称同时验证了另一容器。Servlet应用可以使用WebClient；MVC方法返回Mono也不会把应用变为WebFlux。Boot4同时具备MVC/WebFlux正常模块时优先推导SERVLET，参见应用类型原文。实验仅引入webflux starter，并把服务器绑定127.0.0.1、随机端口。

DispatcherHandler：找首个匹配，再选择适配器

请求进入有序 WebFilter 链后到 DispatcherHandler。它从上下文取并排序 HandlerMapping、HandlerAdapter、HandlerResultHandler。注解方法由 RequestMappingHandlerMapping/RequestMappingHandlerAdapter 处理；函数路由由 RouterFunctionMapping/HandlerFunctionAdapter 处理，结果由 ServerResponseResultHandler 写出。共同的是调度入口，具体适配器和结果类型不同。

原码：Framework v7.0.10，DispatcherHandler#handle，从官方固定tag按原行节选：

原码java

	@Override
	public Mono<Void> handle(ServerWebExchange exchange) {
		if (this.handlerMappings == null) {
			return createNotFoundError();
		}
		if (CorsUtils.isPreFlightRequest(exchange.getRequest())) {
			return handlePreFlight(exchange);
		}
		return Flux.fromIterable(this.handlerMappings)
				.concatMap(mapping -> mapping.getHandler(exchange))
				.next()
				.switchIfEmpty(createNotFoundError())
				.onErrorResume(ex -> handleResultMono(exchange, Mono.error(ex)))
				.flatMap(handler -> handleRequestWith(exchange, handler));
	}

concatMap 按映射顺序查询，next() 取第一个发出的handler；不是并发竞速。全部为空才切到404错误。onErrorResume 把错误送入结果处理路径，具体异常处理器仍决定能否恢复；不能把它理解成所有异常都自动成功。handleRequestWith 内再找支持这个handler的adapter，异步结果交给合适的result handler。错误发生在响应提交之后，无法重新发送状态行，此时要按断流而不是“改成统一500 JSON”处置。

WebFilter 是 Mono<Void> filter(exchange, chain)，应返回 chain.filter(exchange) 组成的链；在过滤器内部自行subscribe会脱离请求生命周期。实验A(order1)、B(order2)进入为A→B，用doOnSuccess观察成功信号为B→A。草稿原用doFinally时实测顺序为A→B，因为它在终止向下游传播后运行；不能把doFinally当作同一个洋葱退出钩子或业务提交钩子；要在提交前加响应头，用响应尚未提交的阶段或beforeCommit。

注解、函数端点与媒体类型

下面是可运行实验中的最小端点（完整imports和启动配置在LabApp）。注解负责绑定路径与返回类型；函数式路由显式组合谓词与响应。两者都必须返回Publisher，框架负责订阅。

java

@RestController
class Api {
    @GetMapping("/api/mono") Mono<String> mono() { return Mono.just("mono-ok"); }
    record Event(String value) { }
    @GetMapping("/api/flux") Flux<Event> flux() {
        return Flux.just("a", "b", "c").map(Event::new);
    }
}
@Bean
RouterFunction<ServerResponse> route() {
    return RouterFunctions.route(RequestPredicates.GET("/fn/echo/{name}"), req ->
        ServerResponse.ok().bodyValue("echo:" + req.pathVariable("name")));
}

Mono<String> 不强制变成JSON字符串。本组合无显式媒体类型且客户端Accept不限时，字符串writer选text/plain；实验断言text/plain;charset=UTF-8。同一Flux<Event>，显式Accept application/json得到对象数组，Accept text/event-stream得到逐项data:{"value":"a"}等SSE事件；没有媒体类型条件，不能用“Flux必然流式”解释网络输出。无界流编码成单一JSON数组可能永远不结束；选择SSE/NDJSON时仍需客户端解析、缓冲上限与断线恢复设计。字符串有特殊writer路径：Flux<String>配Accept application/json实测是原样拼接abc，没有自动JSON引用/数组包装；要JSON结构，应返回DTO/record让JSON encoder处理，或自己承担合法JSON序列化，不能只换Content-Type。

函数路由组合.and(另一条路由)按顺序匹配，实验同路径先声明者返回first。注解路由重叠还涉及路径具体性等比较规则，不能照搬“声明先者总赢”。请求体用@RequestBody Mono<Input>或request.bodyToMono(Input.class)读取一次；解码错误走错误信号，使用@Valid时区别校验目标与方法参数校验。Servlet的HandlerInterceptor、消息转换器或MultipartFile不能原样插到reactive链；选择WebFilter、响应式codecs与Part/FilePart。生产输入必须设codec内存限制并考虑聚合成本，DataBuffer引用计数专课归U49。

实验代码java

package study.webflux;

import java.time.Duration;
import java.util.List;
import java.util.UUID;

import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.context.ApplicationContext;
import org.springframework.http.MediaType;
import org.springframework.test.web.reactive.server.WebTestClient;

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

/**
 * U27组1：注解端点与函数端点同一语义；WebTestClient.bindToApplicationContext（不启真实端口）。
 * 无映射路径的 404 来自 DispatcherHandler 的 switchIfEmpty(createNotFoundError)（源码节选对照）；
 * WebFilter 顺序洋葱断言 + 响应式安全上下文传播（写 Reactor Context，读 ReactiveSecurityContextHolder）。
 */
@SpringBootTest(classes = LabApp.class, properties = "server.address=127.0.0.1")
class AnnotationFunctionEndpointsTest {

    @Autowired
    ApplicationContext context;

    WebTestClient client;

    @BeforeEach
    void bind() {
        // 公开入口要求：bindToApplicationContext 断言（非 mock 单个 controller，走完整 DispatcherHandler 链）
        client = WebTestClient.bindToApplicationContext(context)
                .configureClient().responseTimeout(Duration.ofSeconds(10)).build();
    }

    @Test
    void annotationEndpointReturnsMono() {
        client.get().uri("/api/mono").exchange()
                .expectStatus().isOk()
                .expectHeader().contentTypeCompatibleWith(MediaType.TEXT_PLAIN)
                .expectBody(String.class).isEqualTo("mono-ok");
        System.out.println("U27OBSERVE endpoints annotation mono=mono-ok");
    }

    @Test
    void annotationEndpointFluxIsJsonArray() {
        client.get().uri("/api/flux").accept(MediaType.APPLICATION_JSON).exchange()
                .expectStatus().isOk()
                .expectBody().json("[{\"value\":\"a\"},{\"value\":\"b\"},{\"value\":\"c\"}]");
        System.out.println("U27OBSERVE endpoints annotation fluxJson=[{value:a}, {value:b}, {value:c}]");
    }

    @Test
    void annotationEndpointFluxStreamsAsSse() {
        String body = client.get().uri("/api/flux").accept(MediaType.TEXT_EVENT_STREAM).exchange()
                .expectStatus().isOk()
                .expectHeader().contentTypeCompatibleWith(MediaType.TEXT_EVENT_STREAM)
                .expectBody(String.class).returnResult().getResponseBody();
        assertTrue(body.contains("data:{\"value\":\"a\"}") && body.contains("data:{\"value\":\"c\"}"), "SSE 逐个推送，实测 " + body);
        System.out.println("U27OBSERVE endpoints annotation fluxSse=" + body.replace("\n", "\\n"));
    }

    @Test
    void stringFluxUsesStringEncoderEvenWithJsonMediaType() {
        client.get().uri("/api/strings").accept(MediaType.APPLICATION_JSON).exchange()
                .expectStatus().isOk().expectBody(String.class).isEqualTo("abc");
        System.out.println("U27OBSERVE endpoints stringFluxJsonMedia raw=abc (not a JSON array)");
    }

    @Test
    void functionEndpointSameSemantics() {
        client.get().uri("/fn/echo/webflux").exchange()
                .expectStatus().isOk()
                .expectBody(String.class).isEqualTo("echo:webflux");
        System.out.println("U27OBSERVE endpoints function echo=echo:webflux");
    }

    @Test
    void routerFunctionCompositionOrderFirstWins() {
        // RouterFunction 顺序组合：同一路径先声明的路由先匹配——DispatcherHandler 对 mapping 也是顺序遍历
        String body = client.get().uri("/fn/order").exchange()
                .expectStatus().isOk()
                .expectBody(String.class).returnResult().getResponseBody();
        assertEquals("first", body, "先声明的路由必须获胜，实测 " + body);
        System.out.println("U27OBSERVE endpoints routerOrder winner=first");
    }

    @Test
    void unmappedPathIsDispatcherHandlerNotFound() {
        client.get().uri("/no-such-path").exchange().expectStatus().isNotFound();
        System.out.println("U27OBSERVE endpoints notFound=404 (DispatcherHandler switchIfEmpty createNotFoundError)");
    }

    @Test
    void webFiltersFormOrderedOnion() {
        String key = UUID.randomUUID().toString();
        client.get().uri("/api/mono").header("X-Case", key).exchange()
                .expectStatus().isOk()
                .expectHeader().values("X-Trace", values -> assertEquals(List.of("A", "B"), values, "进入顺序 A→B"))
                .expectBody().consumeWith(result -> { });
        List<String> sequence = LabApp.FILTER_TRACE.get(key);
        assertEquals(List.of("A-in", "B-in", "B-out", "A-out"), sequence,
                "进入 order(1)→order(2)，完成信号按内层先结束后展开：洋葱顺序");
        System.out.println("U27OBSERVE endpoints filterOnion=" + sequence);
    }

    @Test
    void securityContextTravelsInReactorContextNotThreadLocal() {
        String body = client.get().uri("/api/who").exchange()
                .expectStatus().isOk()
                .expectBody(String.class).returnResult().getResponseBody();
        assertTrue(body.contains("ctx=u27-user"), "WebFilter 写入的响应式安全上下文必须在 handler 可读，实测 " + body);
        assertTrue(body.contains("threadLocal=none"), "ThreadLocal SecurityContextHolder 不承载该身份，实测 " + body);
        System.out.println("U27OBSERVE endpoints security " + body);
    }
}

事件循环：不抛异常不代表没有阻塞

实验真实HTTP服务的handler运行在webflux-http-nio-N，客户端响应交付在reactor-http-nio-N；两者来自本组合不同LoopResources，名字不是规范。依赖行为判断应看调度模型与NonBlocking标记，不把某个前缀写成通用运行保证。

在服务端调用Mono.just("x").map(v -> v).block()，进入BlockingSingleSubscriber的NonBlocking检查并抛IllegalStateException，消息包含当前线程。Mono.just("x").block()被MonoJust直接返回值，绕过此检查；Thread.sleep、JDBC和文件阻塞也不会自动被这个检测器阻止。检测不到与可以阻塞是两回事。 机制源码和标量反例见Reactor阻塞隔离。

临时接入阻塞API时，惰性包装并切到有界执行资源：

java

return Mono.fromCallable(() -> blockingLibrary.load(id))
    .subscribeOn(Schedulers.boundedElastic());

不要先load(id)再Mono.just(result)，工作早在切换前发生。boundedElastic是阻塞适配方案之一，也可用受控专用Scheduler；它的队列和线程仍有上限，不能替代下游并发接纳。实验真实sleep120ms在boundedElastic线程，HTTP恢复正常；这不是吞吐基准，也不能保证旧JDBC操作收到cancel后必然停止。

WebClient：订阅、状态、超时与重试

WebClient.builder().baseUrl(url).build()构建客户端，不发请求；返回的冷Mono在被订阅时发起请求，重订阅可能再次执行HTTP副作用。正常应用由框架订阅返回链，只有CLI或测试的非事件循环入口才用有时限的block。

java

Mono<String> body = client.get().uri("/flaky")
    .retrieve().bodyToMono(String.class)
    .retryWhen(Retry.fixedDelay(2, Duration.ofMillis(50))
        .filter(e -> e instanceof WebClientResponseException.ServiceUnavailable))
    .timeout(Duration.ofSeconds(2));

这是使用示例：只对选定503重试，重试2次即最多3次请求，总链2秒包含所有尝试与等待。实验使用受控500、500、200三次响应，明确不设置过滤以观察重订阅数=3；生产不能照抄为所有4xx/5xx/解码错误都重试。timeout放retry里面会给每次新预算，放外面限制总链；连接建立/池等待/响应读超时是连接器参数，语义与总截止时间不同。未配置不等于无限资源，连接池容量与codec缓冲也要检查。

retrieve()默认将4xx/5xx转为WebClientResponseException；可用onStatus细分。exchangeToMono(response -> …)显式检查状态并在回调中消费或释放响应体。实验同一个404，前者抛NotFound，后者读出404:missing-body；别把响应对象返回回调外再期待响应体永远可读。响应体消费结束、取消或错误时的连接回收由连接器处理，使用低层API时仍需正确释放。

同一受控400ms后端上，RestClient的同步调用占住调用线程，WebClient的subscribe先返回，稍后在事件循环交付结果。订阅返回快只是调用线程未等待，不是业务完成；实验用闩锁等待最终响应，避免以“拿到Disposable”冒充成功。

取消跨过HTTP：终止消费不能撤销业务

受控Netty服务每30ms发送一个SSE事件。客户端bodyToFlux(String.class).take(2)收到两个事件完成，取消继续向WebClient上游传播；本HTTP/1.1测试中服务端发送流观察到doOnCancel，等待后产帧计数不再增加，随后新请求成功。这验证真实网络释放链路，区别于只bindToApplicationContext的内存测试。

HTTP没有通用的Reactive Streams REQUEST_N/CANCEL消息。请求量影响本地读取、缓冲和传输流控；断开/停止读取的具体网络动作由连接器及协议决定，不能把观察解释成“客户端request(2)让远端恰好只计算两条”。允许已有在途帧，实验在服务器观察到取消后冻结计数。服务端已提交订单而客户端超时，订单仍可成立；恢复方应带幂等键查询状态，不能凭取消决定再扣一次款。

实验代码java

package study.webflux;

import java.io.IOException;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
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.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

import com.sun.net.httpserver.HttpExchange;
import com.sun.net.httpserver.HttpServer;

import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

import org.springframework.http.MediaType;
import org.springframework.web.client.RestClient;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.reactive.function.client.WebClientResponseException;

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;

/**
 * U27组3：WebClient 真实往返（受控 JDK HttpServer + reactor-netty 流式服务，全部 loopback 随机端口）。
 * 覆盖：真实往返与交付线程、超时取消、retryWhen 重试计数、exchangeToMono 低层控制 vs retrieve 异常、
 * RestClient 同场景对照（阻塞线）、客户端 take(2) 取消沿 HTTP 链传播到服务端发送侧（复用 U26 取消结论）。
 */
class WebClientHttpTest {

    HttpServer backend;
    int backendPort;
    AtomicInteger flakyAttempts = new AtomicInteger();

    WebClient client = WebClient.builder().build();

    @BeforeEach
    void startControlledBackend() throws IOException {
        backend = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
        backend.createContext("/ok", ex -> respond(ex, 200, "backend-ok"));
        backend.createContext("/flaky", ex -> {
            if (flakyAttempts.incrementAndGet() < 3) {
                respond(ex, 500, "fail-" + flakyAttempts.get());
            }
            else {
                respond(ex, 200, "recovered");
            }
        });
        backend.createContext("/slow", ex -> {
            try {
                Thread.sleep(400);
            }
            catch (InterruptedException ignored) {
                Thread.currentThread().interrupt();
            }
            respond(ex, 200, "slow-done");
        });
        backend.createContext("/missing", ex -> respond(ex, 404, "missing-body"));
        backend.start();
        backendPort = backend.getAddress().getPort();
    }

    @AfterEach
    void stopBackend() {
        if (backend != null) backend.stop(0);
    }

    private static void respond(HttpExchange ex, int status, String body) throws IOException {
        byte[] bytes = body.getBytes(StandardCharsets.UTF_8);
        ex.getResponseHeaders().add("Content-Type", "text/plain");
        ex.sendResponseHeaders(status, bytes.length);
        try (OutputStream out = ex.getResponseBody()) {
            out.write(bytes);
        }
    }

    private String uri(String path) {
        return "http://127.0.0.1:" + backendPort + path;
    }

    @Test
    void realRoundtripOnLoopback() {
        long t0 = System.nanoTime();
        String body = client.get().uri(uri("/ok")).retrieve().bodyToMono(String.class).block();
        long elapsedMs = (System.nanoTime() - t0) / 1_000_000;
        assertEquals("backend-ok", body);
        assertTrue(elapsedMs < 5000, "loopback 往返应远小于秒级，实测 " + elapsedMs + "ms");
        System.out.println("U27OBSERVE webclient roundtrip body=backend-ok elapsedMs=" + elapsedMs);
    }

    @Test
    void responseDeliveredOnClientEventLoop() {
        AtomicReference<String> thread = new AtomicReference<>();
        client.get().uri(uri("/ok")).retrieve().bodyToMono(String.class)
                .doOnNext(v -> thread.set(Thread.currentThread().getName())).block();
        assertTrue(thread.get() != null && thread.get().startsWith("reactor-http-nio"),
                "WebClient 响应体在客户端事件循环交付，实测 " + thread.get());
        System.out.println("U27OBSERVE webclient deliverThread=" + thread.get());
    }

    @Test
    void timeoutCancelsBeforeBackendSleepsOut() {
        long t0 = System.nanoTime();
        String result = client.get().uri(uri("/slow")).retrieve().bodyToMono(String.class)
                .timeout(Duration.ofMillis(100))
                .onErrorResume(java.util.concurrent.TimeoutException.class, e -> {
                    // timeout 先 cancel 上游（U26 timeout 取消链），再给出换轨值
                    return Mono.just("timeout-fallback");
                })
                .block();
        long elapsedMs = (System.nanoTime() - t0) / 1_000_000;
        assertEquals("timeout-fallback", result);
        assertTrue(elapsedMs < 400, "超时取消必须先于服务端 400ms 睡眠完成，实测 " + elapsedMs + "ms");
        System.out.println("U27OBSERVE webclient timeout fallbackMs=" + elapsedMs);
    }

    @Test
    void retryWhenCountsAttemptsAtBackend() {
        String body = client.get().uri(uri("/flaky")).retrieve().bodyToMono(String.class)
                .retryWhen(Retry.fixedDelay(2, Duration.ofMillis(50)))
                .block();
        assertEquals("recovered", body);
        assertEquals(3, flakyAttempts.get(), "1次初始 + 2次重试，服务端计数为证");
        System.out.println("U27OBSERVE webclient retry attempts=" + flakyAttempts.get() + " body=recovered");
    }

    @Test
    void exchangeKeepsRawControlWhileRetrieveThrows() {
        // exchange：低层控制——4xx 不抛异常，状态码与响应体都由调用方接管
        String handled = client.get().uri(uri("/missing"))
                .exchangeToMono(resp -> resp.bodyToMono(String.class)
                        .map(body -> resp.statusCode().value() + ":" + body))
                .block();
        assertEquals("404:missing-body", handled);
        // retrieve：同一响应走默认错误路径，抛类型化异常
        assertThrows(WebClientResponseException.NotFound.class, () ->
                client.get().uri(uri("/missing")).retrieve().bodyToMono(String.class).block());
        System.out.println("U27OBSERVE webclient exchange handled=404:missing-body retrieve=WebClientResponseException.NotFound");
    }

    @Test
    void restClientBlocksWhileWebClientFreesCaller() throws Exception {
        // 回指 HTTP 客户端原文：同一受控服务端的两条线
        RestClient rest = RestClient.create();
        long r0 = System.nanoTime();
        String body = rest.get().uri(uri("/slow")).retrieve().body(String.class);
        long restElapsedMs = (System.nanoTime() - r0) / 1_000_000;
        assertEquals("slow-done", body);
        assertTrue(restElapsedMs >= 380, "RestClient 同步阻塞调用线程，实测 " + restElapsedMs + "ms");

        CountDownLatch done = new CountDownLatch(1);
        AtomicReference<String> received = new AtomicReference<>();
        AtomicReference<String> onThread = new AtomicReference<>();
        long w0 = System.nanoTime();
        client.get().uri(uri("/slow")).retrieve().bodyToMono(String.class)
                .subscribe(v -> {
                    received.set(v);
                    onThread.set(Thread.currentThread().getName());
                    done.countDown();
                });
        long afterSubscribeMs = (System.nanoTime() - w0) / 1_000_000;
        assertTrue(afterSubscribeMs < 150, "subscribe 立即返回，调用线程不被占住，实测 " + afterSubscribeMs + "ms");
        assertTrue(done.await(2, TimeUnit.SECONDS), "响应必须在服务端睡眠后到达");
        assertEquals("slow-done", received.get());
        assertTrue(onThread.get().startsWith("reactor-http-nio"), "交付在事件循环线程，实测 " + onThread.get());
        System.out.println("U27OBSERVE webclient blockingLine restClientMs=" + restElapsedMs
                + " webClientSubscribeReturnMs=" + afterSubscribeMs + " webClientDeliverThread=" + onThread.get());
    }

    @Test
    void clientTake2CancelPropagatesToServerSideSend() throws Exception {
        // reactor-netty 受控流式服务：每 30ms 一帧 SSE；客户端 take(2) 收满即自动取消（U26 take 语义）
        CountDownLatch serverCancelled = new CountDownLatch(1);
        AtomicInteger frames = new AtomicInteger();
        var streamingServer = reactor.netty.http.server.HttpServer.create()
                .host("127.0.0.1").port(0)
                .handle((req, res) -> {
                    res.header("Content-Type", MediaType.TEXT_EVENT_STREAM_VALUE);
                    return res.send(Flux.interval(Duration.ofMillis(30))
                            .doOnNext(i -> frames.incrementAndGet())
                            .map(i -> res.alloc().buffer()
                                    .writeBytes(("data:tick-" + i + "\n\n").getBytes(StandardCharsets.UTF_8)))
                            .doOnCancel(serverCancelled::countDown));
                })
                .bindNow();
        try {
            List<String> received = new CopyOnWriteArrayList<>();
            CountDownLatch done = new CountDownLatch(1);
            client.get().uri("http://127.0.0.1:" + streamingServer.port() + "/stream")
                    .retrieve().bodyToFlux(String.class)
                    .take(2) // 收满 2 帧自动向上游 cancel——沿 WebClient/HTTP/服务端发送流传播
                    .subscribe(v -> received.add(v), e -> { }, done::countDown);
            assertTrue(done.await(3, TimeUnit.SECONDS), "take(2) 收满即完成");
            assertEquals(List.of("tick-0", "tick-1"), received);
            assertTrue(serverCancelled.await(3, TimeUnit.SECONDS), "客户端取消必须传播到服务端发送侧 doOnCancel");
            int framesAtCancel = frames.get();
            Thread.sleep(150);
            assertEquals(framesAtCancel, frames.get(), "取消后服务端不再产帧，实测 frames=" + frames.get());
            // 取消被中断的连接后，新的请求依然可用（资源可恢复，服务端仍服务新连接）
            client.get().uri("http://127.0.0.1:" + streamingServer.port() + "/stream")
                    .retrieve().bodyToFlux(String.class).take(2).collectList().block(Duration.ofSeconds(3));
            System.out.println("U27OBSERVE webclient cancelToServer received=2 framesAtCancel=" + framesAtCancel
                    + " serverCancelled=true nextRequestOk=true");
        }
        finally {
            streamingServer.disposeNow();
        }
    }
}

R2DBC：连接资源绑定在订阅上下文

R2DBC是响应式数据库SPI，需要对应驱动；JDBC Connection不能交给R2dbcTransactionManager。实验显式ConnectionFactory、DatabaseClient与R2dbcTransactionManager；Boot自动配置退让条件见集成条件原文。本票不安装连接池，无事务对照每次创建新物理连接；有池时“连接ID不同”的对照不再必然成立。

下面是完整参数语义；默认本机root/空口令只是现有隔离教学环境，其他设置通过U27_MYSQL_HOST/PORT/USER/PASSWORD传入，不写凭据到教材或证据。

java

ConnectionFactory cf = ConnectionFactories.get(ConnectionFactoryOptions.builder()
    .option(ConnectionFactoryOptions.DRIVER, "mysql")
    .option(ConnectionFactoryOptions.HOST, "127.0.0.1")
    .option(ConnectionFactoryOptions.PORT, 3306)
    .option(ConnectionFactoryOptions.USER, System.getenv("U27_MYSQL_USER"))
    .option(ConnectionFactoryOptions.PASSWORD, System.getenv("U27_MYSQL_PASSWORD"))
    .option(ConnectionFactoryOptions.DATABASE, databaseCreatedForThisRun).build());
DatabaseClient db = DatabaseClient.create(cf);
TransactionalOperator tx = TransactionalOperator.create(new R2dbcTransactionManager(cf));
Mono<Void> save = db.sql("INSERT INTO account(name) VALUES (:name)")
    .bind("name", "alice").then()
    .then(db.sql("INSERT INTO account(name) VALUES (:name)").bind("name", "bob").then())
    .as(tx::transactional);

save必须被订阅；then组合顺序并保留同一订阅上下文，手动在事务方法内subscribe会创建另一条订阅。不能用返回void的命令式事务方法去包异步SQL，应该返回Mono/Flux，声明式代理选择ReactiveTransactionManager。实验@EnableTransactionManagement代理服务方法返回Mono，两次插入提交；插入后返回错误信号整体回滚；只调用方法不订阅则无新增行。自调用代理限制仍适用，参见AOP边界。

原码：Framework v7.0.10 TransactionContextManager#currentContext：

原码java

	 */
	public static Mono<TransactionContext> currentContext() {
		return Mono.deferContextual(ctx -> {
			if (ctx.hasKey(TransactionContext.class)) {
				return Mono.just(ctx.get(TransactionContext.class));
			}
			if (ctx.hasKey(TransactionContextHolder.class)) {
				TransactionContextHolder holder = ctx.get(TransactionContextHolder.class);
				if (holder.hasContext()) {
					return Mono.just(holder.currentContext());
				}
			}
			return Mono.error(new NoTransactionInContextException());
		});
	}

入口是deferContextual，先找TransactionContext，再查holder；缺失时返回NoTransaction错误。事务资源表随后按ConnectionFactory绑定连接，DatabaseClient通过ConnectionFactoryUtils获取参与事务的连接。同一线程不是充分条件，同一订阅上下文及资源工厂才是关键。实验在一个tx里执行两次SELECT CONNECTION_ID()得到相同值，外部无池的两次查询不同；这是一条事务资源证据，不证明跨两个数据库自动原子提交。

提交、错误、取消：三条收尾路径

原码：Framework v7.0.10 TransactionalOperatorImpl#execute：

原码java

	@Override
	public <T> Flux<T> execute(TransactionCallback<T> action) throws TransactionException {
		return TransactionContextManager.currentContext().flatMapMany(context ->
			Flux.usingWhen(
				this.transactionManager.getReactiveTransaction(this.transactionDefinition),
				action::doInTransaction,
				this.transactionManager::commit,
				this::rollbackOnException,
				this.transactionManager::rollback)
			.onErrorMap(this::unwrapIfResourceCleanupFailure))
		.contextWrite(TransactionContextManager.getOrCreateContext())
		.contextWrite(TransactionContextManager.getOrCreateContextHolder());
	}

usingWhen取得事务资源后执行action，三个清理回调分别是正常完成→commit，错误→rollbackOnException，取消→rollback；contextWrite准备事务上下文。官方固定版本Cancel Signals明确自5.3起取消导致回滚。这不是只由本次实验偶然推断的行为；但rollback需要异步完成，也可能因连接故障失败。

特别危险的是tx.transactional(flux).take(1)：take向事务Publisher发取消，即使下游正常complete，事务也可能回滚。若业务语义只处理第一条，先把受限工作组装进事务范围，再完整消费事务结果；仍须确认范围中SQL/锁符合需求，不能机械交换操作符就宣布业务正确。

实测正常两条插入增量+2；错误信号回滚增量0；第一条插入已经执行、后面故意等待20秒时取消订阅，独立连接复查增量0。实测窗口是事务体未完成，不是“commit发出后再取消也能撤销提交”。提交确认丢失仍是结果未知，需要持久业务标识查询和恢复；不允许用同一个被取消的订阅观察清理完毕，本实验等待并以独立连接验证。

实验代码java

package study.webflux;

import java.time.Duration;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;


import io.r2dbc.spi.ConnectionFactories;
import io.r2dbc.spi.ConnectionFactory;
import io.r2dbc.spi.ConnectionFactoryOptions;

import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.r2dbc.connection.R2dbcTransactionManager;
import org.springframework.transaction.annotation.EnableTransactionManagement;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.reactive.TransactionalOperator;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

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

/**
 * U27组4：R2DBC 真实库（本机 MySQL 8.4.7，127.0.0.1:3306，r2dbc-mysql 1.4.3 由 Boot4.0.9 BOM 管理）。
 * 覆盖：事务内多语句共享同一连接（事务资源绑定进 Reactor Context）、提交/错误信号回滚的行数观察、
 * @Transactional 声明式边界与"不订阅不执行"、事务中途取消的结局观察（官方文档标明应避免的路径）。
 * 资源：独有数据库 u27_r2dbc_<随机串>，@AfterAll 定向 DROP DATABASE 清理。
 */
class R2dbcTransactionTest {

    static final String DB_NAME = "u27_r2dbc_" + UUID.randomUUID().toString().replace("-", "");
    static boolean created;

    static ConnectionFactory admin;
    static ConnectionFactory lab;
    static DatabaseClient db;
    static TransactionalOperator op;
    static AnnotationConfigApplicationContext annoCtx;
    static String serverVersion;

    @BeforeAll
    static void setup() {
        admin = ConnectionFactories.get(ConnectionFactoryOptions.builder()
                .option(ConnectionFactoryOptions.DRIVER, "mysql")
                .option(ConnectionFactoryOptions.HOST, System.getenv().getOrDefault("U27_MYSQL_HOST", "127.0.0.1"))
                .option(ConnectionFactoryOptions.PORT, Integer.parseInt(System.getenv().getOrDefault("U27_MYSQL_PORT", "3306")))
                .option(ConnectionFactoryOptions.USER, System.getenv().getOrDefault("U27_MYSQL_USER", "root"))
                .option(ConnectionFactoryOptions.PASSWORD, System.getenv().getOrDefault("U27_MYSQL_PASSWORD", ""))
                .option(ConnectionFactoryOptions.DATABASE, "mysql")
                .build());
        // CREATE 成功后才取得所有权；从不删除同名既有库。
        exec(admin, "CREATE DATABASE " + DB_NAME);
        created = true;
        lab = ConnectionFactories.get(ConnectionFactoryOptions.builder()
                .option(ConnectionFactoryOptions.DRIVER, "mysql")
                .option(ConnectionFactoryOptions.HOST, System.getenv().getOrDefault("U27_MYSQL_HOST", "127.0.0.1"))
                .option(ConnectionFactoryOptions.PORT, Integer.parseInt(System.getenv().getOrDefault("U27_MYSQL_PORT", "3306")))
                .option(ConnectionFactoryOptions.USER, System.getenv().getOrDefault("U27_MYSQL_USER", "root"))
                .option(ConnectionFactoryOptions.PASSWORD, System.getenv().getOrDefault("U27_MYSQL_PASSWORD", ""))
                .option(ConnectionFactoryOptions.DATABASE, DB_NAME)
                .build());
        exec(lab, "CREATE TABLE account (id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(64))");
        db = DatabaseClient.create(lab);
        op = TransactionalOperator.create(new R2dbcTransactionManager(lab));
        serverVersion = db.sql("SELECT VERSION() AS v").map((row, meta) -> row.get("v", String.class)).first().block();
        // @Transactional 声明式线：独立的 Spring 上下文（@EnableTransactionManagement + CGLIB 代理）
        annoCtx = new AnnotationConfigApplicationContext(TxConfig.class);
        System.out.println("U27OBSERVE r2dbc-tx serverVersion=" + serverVersion);
    }

    @AfterAll
    static void teardown() {
        if (annoCtx != null) {
            annoCtx.close();
        }
        if (created) {
            exec(admin, "DROP DATABASE " + DB_NAME); // 失败让本次测试失败，不能冒称清理成功
            System.out.println("U27OBSERVE r2dbc-tx cleanup databaseDropped=true");
        }
    }

    static void exec(ConnectionFactory cf, String sql) {
        Mono.usingWhen(cf.create(),
                        conn -> Mono.from(conn.createStatement(sql).execute()).then(),
                        io.r2dbc.spi.Connection::close)
                .block(Duration.ofSeconds(10));
    }

    static long count() {
        return db.sql("SELECT COUNT(*) AS c FROM account").map((row, meta) -> row.get("c", Long.class)).first().block();
    }

    static Mono<Long> connectionId() {
        return db.sql("SELECT CONNECTION_ID() AS cid").map((row, meta) -> row.get("cid", Long.class)).first();
    }

    @Test
    void transactionBindsStatementsToSameConnection() {
        // 事务资源绑定：同一事务里的两条语句拿到同一个物理连接（CONNECTION_ID 相同）——
        // 绑定不是 ThreadLocal，是 Reactor Context（TransactionContextManager 源码对照）
        List<Long> ids = op.execute(status -> connectionId().concatWith(connectionId())).collectList().block();
        assertEquals(2, ids.size());
        assertEquals(ids.get(0), ids.get(1), "事务内多语句必须共享同一连接，实测 " + ids);
        // 对照：无事务时每条语句各自拿连接
        Long outside1 = connectionId().block();
        Long outside2 = connectionId().block();
        assertNotEquals(outside1, outside2, "无事务时两条语句是两个连接，实测 " + outside1 + "/" + outside2);
        System.out.println("U27OBSERVE r2dbc-tx bindInTx=" + ids.get(0) + "," + ids.get(1)
                + " outsideTx=" + outside1 + "!=" + outside2);
    }

    @Test
    void transactionalOperatorCommitsBothInserts() {
        long before = count();
        op.execute(status -> db.sql("INSERT INTO account(name) VALUES ('alice')").then()
                        .then(db.sql("INSERT INTO account(name) VALUES ('bob')").then()))
                .then().block();
        assertEquals(before + 2, count(), "两条插入随事务一起提交");
        System.out.println("U27OBSERVE r2dbc-tx commitRows=" + count());
    }

    @Test
    void errorSignalRollsBackTheWholeTransaction() {
        long before = count();
        assertThrows(IllegalStateException.class, () ->
                op.execute(status -> db.sql("INSERT INTO account(name) VALUES ('carol')").then()
                                .then(Mono.error(new IllegalStateException("boom-tx"))))
                        .then().block());
        assertEquals(before, count(), "错误信号触发回滚：中途插入不得落库");
        System.out.println("U27OBSERVE r2dbc-tx rollbackRows=" + count() + " (before=" + before + ")");
    }

    @Test
    void transactionalAnnotationCommitsAndRollsBack() throws Exception {
        TxService service = annoCtx.getBean(TxService.class);
        long before = count();
        service.insertTwo("dave", "erin").block(); // @Transactional：提交路径
        assertEquals(before + 2, count());
        assertThrows(IllegalStateException.class, () -> service.insertThenFail("frank").block());
        assertEquals(before + 2, count(), "@Transactional 失败方法整体回滚");
        service.insertTwo("gina", "hank"); // 响应式门面：不订阅=不执行
        Thread.sleep(200);
        assertEquals(before + 2, count(), "返回的 Mono 未订阅时不得执行任何 SQL");
        System.out.println("U27OBSERVE r2dbc-tx annotation commit+rollback=ok unsubmittedMonoNoEffect=true");
    }

    @Test
    void cancelDuringTransactionEndsInRollbackForThisVersion() throws Exception {
        // 官方边界：TransactionalOperator javadoc 要求"Transactional Publishers should avoid Subscription
        // cancellation"；但参考手册从5.3起明确取消导致回滚，源码usingWhen第五个回调亦是rollback。
        long before = count();
        CountDownLatch firstInsertSeen = new CountDownLatch(1);

        var d = op.execute(status -> db.sql("INSERT INTO account(name) VALUES ('cancel-a')").then()
                        .doOnSuccess(v -> firstInsertSeen.countDown())
                        .then(Mono.delay(Duration.ofSeconds(20))) // 长尾：给取消留窗口
                        .then(db.sql("INSERT INTO account(name) VALUES ('cancel-b')").then()))
                .subscribe();
        assertTrue(firstInsertSeen.await(5, TimeUnit.SECONDS), "第一条插入必须先于取消发生");
        d.dispose(); // 事务进行中取消订阅
        // 独立连接读事务已结束后的状态；取消不等于同步完成清理。
        Thread.sleep(700);
        long after = count();
        assertEquals(before, after,
                "本版本（spring-tx 7.0.10）：事务体尚未完成时取消 → 未提交的 cancel-a 被回滚");
        System.out.println("U27OBSERVE r2dbc-tx cancel rowsAfterCancel=" + after
                + " (before=" + before + ") outcome=rollback");
    }

    @Configuration
    @EnableTransactionManagement(proxyTargetClass = true)
    static class TxConfig {

        @Bean
        ConnectionFactory connectionFactory() {
            return lab;
        }

        @Bean
        R2dbcTransactionManager transactionManager(ConnectionFactory cf) {
            return new R2dbcTransactionManager(cf);
        }

        @Bean
        DatabaseClient databaseClient(ConnectionFactory cf) {
            return DatabaseClient.create(cf);
        }

        @Bean
        TxService txService() {
            return new TxService();
        }
    }

    static class TxService {

        @Autowired
        DatabaseClient db;

        @Transactional
        public Mono<Void> insertTwo(String a, String b) {
            return db.sql("INSERT INTO account(name) VALUES (:name)").bind("name", a).then()
                    .then(db.sql("INSERT INTO account(name) VALUES (:name)").bind("name", b).then());
        }

        @Transactional
        public Mono<Void> insertThenFail(String name) {
            return db.sql("INSERT INTO account(name) VALUES (:name)").bind("name", name).then()
                    .then(Mono.error(new IllegalStateException("annotation-boom")));
        }
    }
}

响应式身份：Context 是载体，认证仍需验证

ReactiveSecurityContextHolder.getContext()返回Mono，不从SecurityContextHolder的ThreadLocal读取。实验WebFilter将固定合成身份写进chain.filter(exchange).contextWrite(ReactiveSecurityContextHolder.withAuthentication(auth))；handler读到ctx=u27-user，而命令式ThreadLocal没有身份。内存绑定与真实事件循环均验证这点。

这个过滤器是传播探针，没有验证任何密码、token或权限，不能直接部署为认证器。真正安全链需要SecurityWebFilterChain、ReactiveAuthenticationManager/认证转换器、异常翻译与方法/资源授权；响应式方法安全在订阅期使用Reactive上下文，不能把Servlet过滤链顺序和SecurityContextRepository直接照搬。U29/U30负责协议和资源授权，U31负责响应式认证授权实跑及企业扩展可信配置与失败推演，本页不把固定身份写入算作安全通过。ThreadLocal和Context互操作也要显式桥接，跨到任意线程池不自动携带身份。

RSocket：交互模型和协议级取消

RSocket在长连接上复用逻辑stream，TCP/WebSocket是transport，route是Spring消息元数据约定，不能当成HTTP路径。固定RSocket Java1.1.5四种业务交互：

模型	请求/响应	本实验
fire-and-forget	一个请求、无响应payload	独立服务端闩锁确认收到；客户端完成不证明业务持久化
request-response	一个请求、至多一个响应	ping→resp:ping
request-stream	一个请求、多响应	收tick-0/1后take(2)，服务端收到取消并停止产生
request-channel	双向流	a/b/c→echo:a/b/c；两个方向各自遵守请求量与终止规则

另外metadata-push是连接级元数据通知，不携带普通业务响应；SETUP/KEEPALIVE/LEASE/RESUME是连接管理而非第五种请求响应业务成功保证。协议REQUEST_N和CANCEL按stream传递需求与取消，与HTTP的本地取消不同。请求量计的是payload数，实际内存还受每个payload字节数、分片、并发stream与缓冲影响。

原码：RSocket Java 1.1.5 RSocket#requestStream/requestChannel，默认实现交给adapter（未覆写的业务方法不能凭接口名假定可用）：

原码java

  default Flux<Payload> requestStream(Payload payload) {
    return RSocketAdapter.requestStream(payload);
  }

  /**
   * Request-Channel interaction model of {@code RSocket}.
   *
   * @param payloads Stream of request payloads.
   * @return Stream of response payloads.
   */
  default Flux<Payload> requestChannel(Publisher<Payload> payloads) {
    return RSocketAdapter.requestChannel(payloads);
  }

实验raw RSocketServer/Connector绑定loopback随机TCP端口，实现四个方法；接收Payload后读取并释放所有权，发送的Payload交给协议层，不在发送后再次release。取消只终止目标stream；实验再发请求仍成功。选择resume需要双方配置支持并维护有限重放状态、接受窗口和过期处理；resume与重连都不能替代业务幂等、持久化或跨进程灾难恢复。lease限制新请求的接纳，不是对已开始业务的提交确认。本票验证请求/取消，不把断网重连或生产恢复窗口登记为实测。

Spring RSocket：路由、编解码和生命周期

Spring将协议payload与消息方法适配：RSocketMessageHandler收元数据与payload，按@MessageMapping找方法并编解码；RSocketRequester在客户端编码route/data并选择交互模型。默认策略支持String等简单类型；结构化JSON/自定义metadata需双方配兼容codecs、MIME类型与元数据提取，不能认为任意Java对象都自动互通。

java

@Controller
class Messages {
    @MessageMapping("greet") Mono<String> greet(String name) {
        return Mono.just("hello:" + name);
    }
}
RSocketRequester requester = RSocketRequester.builder()
    .rsocketStrategies(strategies).tcp("127.0.0.1", port);
Mono<String> reply = requester.route("greet").data("u27").retrieveMono(String.class);

完整实验为handler配置RSocketStrategies、Spring上下文和真实TCP listener，requester订阅触发连接与交互；不要每个请求都创建一条新长连接。实测路由返回hello:u27，未知路由返回ApplicationErrorException，stream take(2)触发server cancel，随后同连接greet仍可用。应用拥有requester、listener与context，finally逐个dispose/close并等待listener关闭。@ConnectMapping适合SETUP阶段初始化/校验，不能替代每条消息授权；连接时认证后仍需考虑撤销与租户隔离，归U31。

实验代码java

package study.webflux;

import java.time.Duration;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import io.rsocket.core.RSocketServer;
import io.rsocket.transport.netty.server.TcpServerTransport;
import org.junit.jupiter.api.Test;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.messaging.rsocket.RSocketStrategies;
import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler;
import org.springframework.stereotype.Controller;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import static org.junit.jupiter.api.Assertions.*;

/** Spring 消息路由、编解码和取消；真实 TCP。安全认证另归 U31，本类没有认证器。 */
class SpringRSocketTest {
    @Configuration
    static class Config {
        @Bean RSocketStrategies strategies() { return RSocketStrategies.builder().build(); }
        @Bean RSocketMessageHandler handler(RSocketStrategies strategies) {
            var handler = new RSocketMessageHandler();
            handler.setRSocketStrategies(strategies);
            return handler;
        }
        @Bean Messages messages() { return new Messages(); }
    }
    @Controller
    static class Messages {
        final CountDownLatch cancelled = new CountDownLatch(1);
        @MessageMapping("greet") Mono<String> greet(String name) { return Mono.just("hello:" + name); }
        @MessageMapping("ticks") Flux<String> ticks() {
            return Flux.interval(Duration.ofMillis(30)).map(i -> "tick:" + i).doOnCancel(cancelled::countDown);
        }
    }
    @Test
    void springRouteCodecsCancelAndRecovery() throws Exception {
        try (var context = new AnnotationConfigApplicationContext(Config.class)) {
            var server = RSocketServer.create(context.getBean(RSocketMessageHandler.class).responder())
                    .bindNow(TcpServerTransport.create("127.0.0.1", 0));
            RSocketRequester requester = null;
            try {
                requester = RSocketRequester.builder().rsocketStrategies(context.getBean(RSocketStrategies.class))
                        .tcp("127.0.0.1", server.address().getPort());
                assertEquals("hello:u27", requester.route("greet").data("u27").retrieveMono(String.class).block(Duration.ofSeconds(5)));
                var active = requester;
                assertThrows(io.rsocket.exceptions.ApplicationErrorException.class, () -> active.route("missing")
                        .data("x").retrieveMono(String.class).block(Duration.ofSeconds(5)));
                var received = requester.route("ticks").retrieveFlux(String.class).take(2).collectList().block(Duration.ofSeconds(5));
                assertEquals(List.of("tick:0", "tick:1"), received);
                assertTrue(context.getBean(Messages.class).cancelled.await(3, TimeUnit.SECONDS));
                assertEquals("hello:again", requester.route("greet").data("again").retrieveMono(String.class).block(Duration.ofSeconds(5)));
                System.out.println("U27OBSERVE spring-rsocket route=hello:u27 missing=ApplicationErrorException cancel=true recovery=hello:again");
            } finally {
                if (requester != null) requester.dispose();
                server.dispose();
                server.onClose().block(Duration.ofSeconds(5));
            }
        }
    }
}

选择与排障：省掉等待线程后，瓶颈去了哪里

假设窗口10秒，稳定流量1000请求/秒、每请求下游等待200ms，约有200个并行等待（Little定律，只对这个稳定窗口的模型成立）。若数据库只允许40个活跃操作，两种服务模型都要限并发；改用事件循环不会把数据库容量变成200。若每个在途请求保留64KiB数据，200个约12.5MiB；如果无界收集到1万项，则至少625MiB，尚不含对象/网络/直接内存。

两个合理选择：已有事务JDBC、负载能用平台/虚拟线程满足的CRUD服务，保留MVC配有界连接池和接纳，降低异步事务复杂度；端到端非阻塞驱动、长连接多且要流式处理时，选择WebFlux/R2DBC并限制flatMap并发、连接、请求体和响应体缓冲。临时boundedElastic适配遗留库能推进迁移，但多一层排队、取消不可撤销和上下文桥接成本。RSocket适合受控两端的多路流；公网兼容/代理运维优先时可选HTTP/SSE，接受单向流与自行恢复的代价。

排障按证据依次检查：应用类型和依赖→线程名及堆栈（有没有sleep/JDBC/block）→每层队列/连接池等待→媒体类型和codec缓冲→订阅/取消日志→独立连接查询事务终态。HTTP超时由调用方发起状态查询，数据库提交未知由业务恢复方凭唯一键裁决，断流续传由流协议/游标拥有者负责，不能交给框架“自动重试所有操作”。升级驱动或Framework后复验事务取消、连接释放和协议路由；增长超过连接/内存预算而不是某个组件流行时，再考虑架构演进。

来源与可复现边界

Framework v7.0.10 WebFlux官方目录：DispatcherHandler、注解/函数模型、URI、codecs、WebClient、测试与WebSocket；本页承担U27服务/客户端分段，相关内容去向见docs/audits/java-depth-u27-scope.md。
响应式事务固定手册、上述固定源码：规范/实现与本机运行分层。
Spring RSocket固定章节、RSocket协议、RSocket Java1.1.5：协议和Spring路由分层，查询2026-10-26。
实验入口与实测记录：局部绑定、真实HTTP、真实MySQL、真实TCP分别记录，不外推TLS、生产吞吐、跨地域可靠性或全部数据库。高风险事务与取消结论复核见docs/audits/java-depth-u27-verification.md。

闭卷验收

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

WebFlux、R2DBC 与 RSocket：请求、事务和取消 ​

Servlet 与 Reactive：先认清运行边界 ​

DispatcherHandler：找首个匹配，再选择适配器 ​

注解、函数端点与媒体类型 ​

事件循环：不抛异常不代表没有阻塞 ​

WebClient：订阅、状态、超时与重试 ​

取消跨过HTTP：终止消费不能撤销业务 ​

R2DBC：连接资源绑定在订阅上下文 ​

提交、错误、取消：三条收尾路径 ​

响应式身份：Context 是载体，认证仍需验证 ​

RSocket：交互模型和协议级取消 ​

Spring RSocket：路由、编解码和生命周期 ​

选择与排障：省掉等待线程后，瓶颈去了哪里 ​

来源与可复现边界 ​

闭卷验收 ​

WebFlux、R2DBC 与 RSocket：请求、事务和取消

Servlet 与 Reactive：先认清运行边界

DispatcherHandler：找首个匹配，再选择适配器

注解、函数端点与媒体类型

事件循环：不抛异常不代表没有阻塞

WebClient：订阅、状态、超时与重试

取消跨过HTTP：终止消费不能撤销业务

R2DBC：连接资源绑定在订阅上下文

提交、错误、取消：三条收尾路径

响应式身份：Context 是载体，认证仍需验证

RSocket：交互模型和协议级取消

Spring RSocket：路由、编解码和生命周期

选择与排障：省掉等待线程后，瓶颈去了哪里

来源与可复现边界

闭卷验收