阿男聊技术

Project Reactor vs Tokio: Async Foundations and Spring AI StreamAdvisor

中文标题：Project Reactor 与 Tokio：异步底座、Linux 内核共性，以及 Spring AI 流式 Advisor

如果你在 JVM 生态里写 Spring AI 流式聊天，在 Rust 里写 Tokio 网络服务，很容易把两套 API 当成「两个平行世界」。它们确实不等价：一个是 Reactive Streams 库，一个是完整 async runtime。但在 Linux 上，两者最终都会落到同一组内核设施——epoll 事件通知、hrtimer 定时精度、futex 阻塞唤醒，以及 schedule() 调度。本文把这三层（用户态库 / runtime、内核 syscall、Spring AI 应用）串成一条可读链路，并说明 StreamAdvisor 如何用 Flux 承载 LLM token 流——不依赖 WebFlux。

If you stream LLM responses with Spring AI on the JVM and build network services with Tokio in Rust, the two APIs can feel like parallel universes. They are not equivalent: one is a Reactive Streams library, the other a full async runtime. On Linux, however, both converge on the same kernel facilities—epoll for I/O readiness, hrtimer for timed waits, futex for block/wake, and schedule() for thread scheduling. This post connects those three layers (userspace library/runtime, kernel syscalls, Spring AI application code) into one readable path, and shows how StreamAdvisor carries LLM token streams via Flux—without WebFlux.

全文脉络：

1. 分层地图 — Reactor + Netty ≈ Tokio + mio；库与 runtime 的边界
2. Linux 内核共性底座 — epoll、hrtimer、futex、scheduler；与已有 Tokio / 内核系列文互链
3. Project Reactor — Reactive Streams、Flux/Mono、背压
4. Tokio — driver + scheduler + timer；PollEvented 与 mio
5. 设计相似与不等价 — 「响应式」三层含义；非阻塞、组合流水线、调度、流控
6. Spring AI 落点 — CallAdvisor vs StreamAdvisor 双链；WebClient → Flux
7. 逐块与聚合 — chunk、SSE、MessageAggregator

1. 分层地图 / Layering: Library vs Runtime

Reactor 常被说成「JVM 的响应式编程库」，Tokio 则是「Rust 的异步运行时」。更精确地说：reactor-core 只定义 Publisher/Subscriber 与 Flux/Mono 组合子；真正碰网卡、调 epoll_wait 的是 Reactor Netty（Spring WebClient 默认传输）。Tokio 则把 I/O driver（mio）、scheduler、timer 打包进同一个 Runtime¹。因此对比时应写成「Reactor 生态 + Netty」对「Tokio 全家桶」，而不是 reactor-core 对 tokio 一行 API。

Reactor is often described as the JVM reactive library, while Tokio is Rust’s async runtime. More precisely: reactor-core only defines Publisher/Subscriber and Flux/Mono operators; the code that touches the NIC and calls epoll_wait lives in Reactor Netty (Spring WebClient’s default transport). Tokio bundles I/O driver (mio), scheduler, and timer into one Runtime¹. Compare “Reactor ecosystem + Netty” against “Tokio as a whole,” not reactor-core line-by-line against tokio.

flowchart TB
  subgraph jvm [JVM_Stack]
    AppJ[SpringAI_ChatClient]
    FluxJ[reactor_core_Flux]
    NettyJ[Reactor_Netty_EpollEventLoop]
    AppJ --> FluxJ
    FluxJ --> NettyJ
  end

  subgraph rust [Rust_Stack]
    AppR[Tokio_App]
    FutR[Future_poll]
    TokioR[Tokio_Runtime]
    MioR[mio_Poll]
    AppR --> FutR
    FutR --> TokioR
    TokioR --> MioR
  end

  subgraph kernel [Linux_Kernel]
    Epoll[epoll_wait]
    Hrtimer[hrtimer]
    Futex[futex]
    Sched[schedule]
  end

  NettyJ --> Epoll
  MioR --> Epoll
  NettyJ --> Hrtimer
  MioR --> Hrtimer
  TokioR --> Futex
  Epoll --> Sched

2. Linux 内核共性底座 / Linux Kernel Common Ground

Reactor Netty 与 Tokio mio 在 Linux 上共享 epoll 事件环。用户态调用 epoll_wait(2) 进入内核 fs/eventpoll.c：SYSCALL_DEFINE4(epoll_wait) → do_epoll_wait → ep_poll²。当 socket 有数据到达，网络栈通过 sk_wake_async 触发已注册的 wait queue 回调 ep_poll_callback，把就绪 fd 链入 epoll ready list，阻塞在 epoll_wait 上的线程被唤醒。Tokio 系列文³已跟踪 park_timeout → epoll_wait；本文补充 JVM 侧同样落在 epoll——Spring AI DeepSeekApi 的 WebClient 经 Reactor Netty 走同一条 syscall。

On Linux, Reactor Netty and Tokio mio share the epoll event loop. Userspace epoll_wait(2) enters the kernel via fs/eventpoll.c: SYSCALL_DEFINE4(epoll_wait) → do_epoll_wait → ep_poll². When a TCP segment arrives, the network stack calls sk_wake_async, which fires the registered wait-queue callback ep_poll_callback, links the ready fd into the epoll ready list, and wakes the thread blocked in epoll_wait. The Tokio series³ already traced park_timeout → epoll_wait; here we add that the JVM side lands on the same epoll path—Spring AI DeepSeekApi’s WebClient goes through Reactor Netty and the same syscall.

// fs/eventpoll.c — kernel side of epoll_wait(2)
static int do_epoll_wait(int epfd, struct epoll_event __user *events,
                         int maxevents, struct timespec64 *to)
{
    // ...
    return ep_poll(ep, events, maxevents, to);
}

// fs/eventpoll.c — wait-queue callback when a monitored fd becomes ready
static int ep_poll_callback(wait_queue_entry_t *wait, unsigned mode, int sync, void *key)
{
    // ... link epitem into ready list, wake epoll waiter ...
}

定时精度：Tokio 时间轮只向内核注册「最近 deadline」，epoll_wait(timeout) 顺带睡眠；ep_poll 把 timeout 转为 ktime_t，到期由 hrtimer 中断唤醒（详见 hrtimer 纳秒精度分析⁴）。Reactor Netty 的 HashedWheelTimer 与 scheduler delay 在用户态批量收割，底层 timed wait 同样依赖内核 hrtimer。

Timed waits: Tokio’s timer wheel registers only the nearest deadline and piggybacks on epoll_wait(timeout); ep_poll converts the timeout to ktime_t, and hrtimer interrupts wake the waiter when it fires (see hrtimer nanosecond precision⁴). Reactor Netty’s HashedWheelTimer and scheduler delays harvest expirations in userspace; the underlying timed wait still relies on kernel hrtimer.

阻塞与唤醒：worker 线程「无任务可跑」时的 park，以及 JVM LockSupport.park/unpark、JUC AbstractQueuedSynchronizer，慢路径都落到 kernel/futex/ 的 futex_wait / futex_wake（详见 用户态锁与 futex⁵）。异步运行时的协作式调度发生在用户态；内核只在 I/O 未就绪或显式阻塞时通过 schedule() 接管线程。

Block and wake: when a Tokio worker has nothing to poll, park; JVM LockSupport.park/unpark and JUC AbstractQueuedSynchronizer slow paths all reach kernel/futex/ via futex_wait / futex_wake (see userspace locks and futex⁵). Cooperative async scheduling happens in userspace; the kernel’s schedule() takes over only when I/O is not ready or a thread explicitly blocks.

sequenceDiagram
    participant App as UserApp
    participant RT as Tokio_or_ReactorNetty
    participant Kernel as LinuxKernel
    participant NIC as NetworkStack

    App->>RT: poll_or_subscribe
    RT->>Kernel: epoll_wait(timeout)
    Note over Kernel: ep_poll blocks via wait_queue
    NIC->>Kernel: TCP segment arrives
    Kernel->>Kernel: sk_wake_async then ep_poll_callback
    Kernel-->>RT: epoll_wait returns readable
    RT-->>App: wake task or emit Flux chunk

边界说明：macOS/BSD 上 Tokio mio 使用 kqueue，非 epoll；Tokio 可选 io-uring feature⁶ 非默认路径。reactor-core 本身不含 epoll——汇聚点在 Reactor Netty / WebClient 传输层。

Scope notes: on macOS/BSD, Tokio mio uses kqueue, not epoll; Tokio’s optional io-uring feature⁶ is not the default path. reactor-core itself has no epoll—the convergence point is Reactor Netty / WebClient transport.

3. Project Reactor 与 Reactive Streams / Project Reactor and Reactive Streams

Project Reactor 实现 Reactive Streams 规范：Publisher 发布数据、Subscriber 消费，并通过 Subscription.request(n) 做背压（backpressure）。Flux<T> 表示 0..N 个元素的异步序列，Mono<T> 表示 0..1。组合子（map、flatMap、windowUntil）在用户态构建流水线，不绑定特定 I/O 模型——这正是 Spring AI 能在 非 WebFlux 应用里用 Flux 流式读 DeepSeek SSE 的原因。

Project Reactor implements the Reactive Streams spec: a Publisher emits, a Subscriber consumes, and Subscription.request(n) provides backpressure. Flux<T> is an asynchronous 0..N sequence; Mono<T> is 0..1. Operators (map, flatMap, windowUntil) build pipelines in userspace without tying to a specific I/O model—which is why Spring AI can stream DeepSeek SSE via Flux in a non-WebFlux app.

// reactor-core/src/main/java/reactor/core/publisher/Flux.java
public abstract class Flux<T> implements CorePublisher<T> {
    // Reactive Streams Publisher with rich operator surface
}

WebFlux 基于 Reactor，但 StreamAdvisor 只依赖 reactor-core 的 Flux，不要求你把 Controller 写成 Mono<ServerResponse>。HTTP 传输由 WebClient + Reactor Netty 完成；业务 Advisor 链在 Flux 层组合。

WebFlux is built on Reactor, but StreamAdvisor needs only reactor-core’s Flux—you do not have to write controllers as Mono<ServerResponse>. HTTP transport is WebClient + Reactor Netty; the Advisor chain composes at the Flux layer.

4. Tokio 运行时 / Tokio Runtime

Tokio 模块文档把 runtime 服务拆成三件套：I/O event loop（driver）、scheduler、timer¹。Future::poll 是 Rust async 的协作式入口；I/O 类型通过 PollEvented 把 mio::event::Source 注册到 reactor，在 readiness 变化时被唤醒⁷。与 Reactor 的 subscribe / request 不同，Tokio 用 Waker 驱动 task 重新入队——语义不同，但「等 I/O → 被 epoll 叫醒 → 继续推进」的内核路径一致。

Tokio’s module docs split runtime services into three pieces: I/O event loop (driver), scheduler, and timer¹. Future::poll is Rust async’s cooperative entry point; I/O types register mio::event::Source with the reactor via PollEvented and resume when readiness changes⁷. Unlike Reactor’s subscribe / request, Tokio uses Waker to re-queue tasks—different semantics, same kernel path: wait for I/O → epoll wakes the thread → progress resumes.

// tokio/src/io/poll_evented.rs
// PollEvented wraps mio::event::Source and ties it to the Tokio reactor
use mio::event::Source;

入门可读 Rust async/await 与 Tokio runtime⁸；I/O 驱动细节见 Tokio I/O driver 与 mio³。

For an introduction, see Rust async/await and Tokio runtime⁸; I/O driver details are in Tokio I/O driver and mio³.

5. 设计相似与不等价 / Conceptual Parallels (and Non-Equivalence)

5.1 厘清「响应式」/ Disambiguating “Reactive”

日常讨论里「响应式」至少有三层含义，混用会导致「Tokio 算不算响应式」争论各说各话。建议分开看：

In everyday discussion, “reactive” has at least three layers of meaning. Conflating them is why debates like “is Tokio reactive?” go in circles. Keep them separate:

Reactive Manifesto 把 Message Driven 写得很明确：系统靠异步消息传递建立组件边界，通过监控消息队列并在必要时施加 back-pressure 来做负载与流控⁹。这是系统架构层的「响应式」——不要求你必须实现 Reactive Streams 接口。

The Reactive Manifesto states that Message Driven systems use asynchronous message-passing between components and apply back-pressure by shaping and monitoring message queues when necessary⁹. This is an architecture-level notion of “reactive”—it does not require implementing the Reactive Streams interfaces.

Reactive Streams 规范则更进一步：它定义的是跨异步边界（线程、线程池）传递数据流时，接收方不得被迫缓冲任意数量的数据；背压通信本身也必须是非阻塞、异步的¹⁰。JDK 9 起 java.util.concurrent.Flow 与 Reactive Streams 语义 1:1 等价¹⁰。Project Reactor README 的定位正是：「Non-Blocking Reactive Streams Foundation for the JVM」，并实现 Reactive Extensions 风格的 API¹¹。

Reactive Streams goes further: it standardizes data flow across async boundaries so the receiver is not forced to buffer unbounded data, and back-pressure signaling itself must be fully non-blocking and asynchronous¹⁰. Since JDK 9, java.util.concurrent.Flow is semantically 1:1 equivalent to Reactive Streams¹⁰. The Project Reactor README describes itself as a 「Non-Blocking Reactive Streams Foundation for the JVM」 with a Reactive Extensions–inspired API¹¹.

Tokio 算不算「响应式」？ 精确回答要分层：

• Reactive Systems 意义下：是。 事件驱动、异步消息（mpsc/oneshot）、非阻塞 I/O，符合 Message Driven 与 Responsive 的方向⁹¹。
• Reactive Streams 意义下：不是。 Tokio 没有 Publisher/Subscriber 契约；背压通过 bounded channel 等原语各自实现——官方 mpsc 文档写明：channel 满时 send 会等待，「In other words, the channel provides backpressure」¹²，但这是 API 级流控，不是 Subscription.request(n) 标准化协议。
• 日常口语「响应式 = 异步非阻塞」：是。 与 Reactor 在理念上相近，在规范与抽象上不等价。

Is Tokio “reactive”? It depends which layer you mean:

• Reactive Systems sense: yes. Event-driven design, async messaging (mpsc/oneshot), non-blocking I/O align with Message Driven and Responsive goals⁹¹.
• Reactive Streams sense: no. Tokio has no Publisher/Subscriber contract; flow control uses primitives like bounded channels—the official mpsc docs state that when full, send waits: 「In other words, the channel provides backpressure」¹²—but this is API-level flow control, not the standardized Subscription.request(n) protocol.
• Colloquial “reactive = async non-blocking”: yes. Conceptually close to Reactor; not equivalent in spec or abstraction.

flowchart LR
  subgraph broad [Reactive_Systems_Manifesto]
    RS[Responsive_Resilient_Elastic_MessageDriven]
  end

  subgraph spec [Reactive_Streams_Spec]
    Pub[Publisher]
    Sub[Subscriber]
    Req[Subscription_request_n]
    Pub --> Sub
    Sub --> Req
  end

  subgraph impl [Implementations]
    Reactor[Project_Reactor_Flux]
    Tokio[Tokio_Future_mpsc]
  end

  broad --> Reactor
  broad --> Tokio
  spec --> Reactor
  Tokio -.->|no_spec_impl| spec

5.2 对照总表 / Comparison Table

相似点：非阻塞、事件驱动、可组合流水线、可把阻塞工作 offload 到专用线程池；广义上都可归入 Reactive Manifesto 倡导的 Message Driven 方向⁹。不等价点：Tokio 是完整 runtime，不实现 Reactive Streams；Reactor 是流库并实现了该规范¹⁰¹¹，I/O 与线程调度要另配 Netty / Schedulers。把 Spring AI 的 Flux 说成「WebFlux 流式」会误导——准确说法是 Reactive Streams 流式（规范层），HTTP 层恰好用了 Reactor Netty。

Similarities: non-blocking, event-driven, composable pipelines, offload blocking work to dedicated pools; in the broad sense, both align with the Message Driven direction of the Reactive Manifesto⁹. Differences: Tokio is a full runtime that does not implement Reactive Streams; Reactor is a stream library that does¹⁰¹¹—I/O and scheduling need Netty / Schedulers. Calling Spring AI’s Flux “WebFlux streaming” is misleading; the precise term is Reactive Streams streaming (spec layer), with Reactor Netty at the HTTP layer.

6. Spring AI 落点：StreamAdvisor 与双链 / Spring AI: StreamAdvisor and Dual Chains

Spring AI 2.0 的 Advisor 链把 call 与 stream 分成两条并行链：DefaultAroundAdvisorChain 维护 callAdvisors 与 streamAdvisors 两个 Deque，.call() 走 nextCall()，.stream() 走 nextStream()，注册可合并、执行不混合¹³¹⁴。StreamAdvisor 接口返回 Flux<ChatClientResponse>¹⁵；链尾由 ChatModelStreamAdvisor 调用 ChatModel.stream()。

Spring AI 2.0 splits call and stream into two parallel chains: DefaultAroundAdvisorChain keeps separate callAdvisors and streamAdvisors deques; .call() uses nextCall(), .stream() uses nextStream()—registered together, never mixed at execution¹³¹⁴. The StreamAdvisor interface returns Flux<ChatClientResponse>¹⁵; the chain terminates with ChatModelStreamAdvisor calling ChatModel.stream().

flowchart LR
  subgraph callPath [Call_Path]
    CC[ChatClient.call]
    NC[nextCall]
    CA[CallAdvisors]
    CMA[ChatModelCallAdvisor]
    CC --> NC --> CA --> CMA
  end

  subgraph streamPath [Stream_Path]
    CS[ChatClient.stream]
    NS[nextStream]
    SA[StreamAdvisors]
    CMS[ChatModelStreamAdvisor]
    CS --> NS --> SA --> CMS
  end

DeepSeek 流式 HTTP 在 DeepSeekApi.chatCompletionStream：WebClient.post() → bodyToFlux(String.class) 解析 SSE，再 map 为 ChatCompletionChunk¹⁶。每个 chunk 通常是一个 response token 片段（不是把用户请求拆块发送）。DefaultChatClient 的 stream 路径用 advisorChain.getStreamAdvisors() 并设 stream(true) 做观测¹⁷。

DeepSeek streaming HTTP lives in DeepSeekApi.chatCompletionStream: WebClient.post() → bodyToFlux(String.class) for SSE, then map to ChatCompletionChunk¹⁶. Each chunk is usually a response token fragment (not splitting the user request). DefaultChatClient’s stream path uses advisorChain.getStreamAdvisors() with stream(true) for observation¹⁷.

// models/spring-ai-deepseek/.../DeepSeekApi.java
return this.webClient.post()
    .uri(this.getEndpoint(chatRequest))
    .body(Mono.just(chatRequest), ChatCompletionRequest.class)
    .retrieve()
    .bodyToFlux(String.class)
    .takeUntil(SSE_DONE_PREDICATE)
    .mapNotNull(content -> jsonHelper.fromJson(content, ChatCompletionChunk.class));

与 Spring AI 2 订票 Demo 对照：该 Demo 只用 .call() 同步路径；流式场景见 springai_demo ADVISOR_API.md¹⁸。

Compared with the Spring AI 2 booking demo: that demo uses only the synchronous .call() path; streaming scenarios are documented in springai_demo ADVISOR_API.md¹⁸.

sequenceDiagram
    participant Client as ChatClient
    participant Chain as StreamAdvisorChain
    participant Model as DeepSeekChatModel
    participant API as DeepSeekApi
    participant WC as WebClient
    participant Netty as ReactorNetty
    participant Kern as epoll_wait

    Client->>Chain: stream(prompt)
    Chain->>Model: adviseStream
    Model->>API: chatCompletionStream
    API->>WC: POST SSE
    WC->>Netty: async read
    Netty->>Kern: epoll_wait
    Kern-->>Netty: readable
    Netty-->>API: Flux String chunk
    API-->>Model: Flux ChatCompletionChunk
    Model-->>Chain: Flux ChatResponse
    Chain-->>Client: Flux ChatClientResponse

7. 逐块与聚合 / Chunks vs Aggregation

流式 chunk 在链路上可能不完整：DeepSeekApi 对 tool-calling 用 windowUntil + reduce 合并¹⁶；MessageAggregator 把多个 ChatResponse chunk 聚合成完整 AssistantMessage¹⁹；DefaultAroundAdvisorChain.nextStream 还通过 ChatClientMessageAggregator 做链级聚合¹³。业务可以 逐块消费 Flux（例如打字机 UI），也可以在 Advisor after 阶段等待聚合结果——Tool Calling 通常需要后者。

Streaming chunks may be incomplete along the pipeline: DeepSeekApi merges tool-calling fragments with windowUntil + reduce¹⁶; MessageAggregator folds multiple ChatResponse chunks into one AssistantMessage¹⁹; DefaultAroundAdvisorChain.nextStream also aggregates via ChatClientMessageAggregator¹³. Applications may consume Flux chunk-by-chunk (typewriter UI) or wait for aggregated output in an Advisor after hook—Tool Calling usually needs the latter.

// spring-ai-model/.../MessageAggregator.java
/**
 * Helper that for streaming chat responses, aggregate the chat response messages
 * into a single AssistantMessage.
 */
public Flux<ChatResponse> aggregate(Flux<ChatResponse> fluxChatResponse,
        Consumer<ChatResponse> onAggregationComplete) { ... }

参考文献 / References

• 上一篇 Spring AI 2 实战：订票聊天 Demo 的架构、ReAct 与 ToolCallingAdvisor 链