阿男聊技术

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

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

源码钉在本地核对过的 commit：reactor-core 95957bd8d、tokio e77885a4、spring-ai d71200317；内核引用本机 linux 树 6d35786de281 的 fs/eventpoll.c。

1. 分层地图：库 vs 运行时

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。

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 内核共性：epoll

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 链进 ready list，唤醒卡在 epoll_wait 上的线程。Tokio 系列文已跟踪 park_timeout → epoll_wait³；这里补一句：JVM 侧经 Reactor Netty 也落在同一条 syscall。

// fs/eventpoll.c L2442–（内核侧 epoll_wait）
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 L1249–
static int ep_poll_callback(wait_queue_entry_t *wait, unsigned mode, int sync, void *key)
{
    struct epitem *epi = ep_item_from_wait(wait);
    struct eventpoll *ep = epi->ep;
    // ... 把 epitem 链入 ready list，唤醒 epoll waiter ...
}

定时：Tokio 时间轮只向内核登记「最近 deadline」，epoll_wait(timeout) 顺带睡；到期靠 hrtimer 叫醒（见 hrtimer 分析⁴）。Reactor Netty 的 HashedWheelTimer 在用户态批量收割，底层 timed wait 同样依赖 hrtimer。

阻塞：worker「无事可做」时的 park，以及 JVM LockSupport.park，慢路径落到 futex_wait / futex_wake（见 futex 文⁵）。协作式调度在用户态；内核只在 I/O 未就绪或显式阻塞时 schedule()。

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 上 mio 用 kqueue；Tokio 的 io-uring feature⁶ 非默认。reactor-core 本身不含 epoll——汇聚点在 Reactor Netty / WebClient 传输层。

3. Project Reactor：订阅与背压

Project Reactor 实现 Reactive Streams：Publisher 发、Subscriber 收，用 Subscription.request(n) 做背压。Flux<T> 是 0..N，Mono<T> 是 0..1。组合子在用户态搭流水线，不绑死某一种 I/O——所以 Spring AI 能在 非 WebFlux 应用里用 Flux 读 SSE。

订阅入口在 Flux.subscribe：先把外部 Subscriber 收成 CoreSubscriber，再沿 operator 链下钻，最后落到具体 Publisher⁷：

// reactor-core/.../Flux.java L8859–8894（95957bd8d）
@Override
public final void subscribe(Subscriber<? super T> actual) {
    CorePublisher publisher = Operators.onLastAssembly(this);
    CoreSubscriber subscriber = Operators.toCoreSubscriber(actual);
    try {
        if (publisher instanceof OptimizableOperator) {
            OptimizableOperator operator = (OptimizableOperator) publisher;
            while (true) {
                subscriber = operator.subscribeOrReturn(subscriber);
                if (subscriber == null) {
                    return;
                }
                // ... 继续沿 OptimizableOperator 链 ...
            }
        }
        subscriber = Operators.restoreContextOnSubscriberIfPublisherNonInternal(publisher, subscriber);
        publisher.subscribe(subscriber);
    }
    catch (Throwable e) {
        Operators.reportThrowInSubscribe(subscriber, e);
    }
}

日常写 flux.subscribe(x -> …) 时，默认 LambdaSubscriber 在 onSubscribe 里若没自定义 subscription 回调，会直接 request(Long.MAX_VALUE)——也就是「无界请求」，背压靠下游自己消化：

// reactor-core/.../LambdaSubscriber.java L109–125（95957bd8d）
public void onSubscribe(Subscription s) {
    if (Operators.validate(subscription, s)) {
        this.subscription = s;
        if (subscriptionConsumer != null) {
            subscriptionConsumer.accept(s);
        }
        else {
            s.request(Long.MAX_VALUE);
        }
    }
}

需要有界拉取时，用 BaseSubscriber 显式 request(n)：

// reactor-core/.../BaseSubscriber.java L212–227（95957bd8d）
@Override
public final void request(long n) {
    if (Operators.validate(n)) {
        Subscription s = this.subscription;
        if (s != null) {
            s.request(n);
        }
    }
}

public final void requestUnbounded() {
    request(Long.MAX_VALUE);
}

Operators.validate(n) 拒绝 n <= 0；非 CoreSubscriber 会被包成 StrictSubscriber，保证 onSubscribe 里先设状态再 request⁷。

WebFlux 基于 Reactor，但 StreamAdvisor 只依赖 reactor-core 的 Flux，不必把 Controller 写成 Mono<ServerResponse>。HTTP 仍是 WebClient + Reactor Netty；Advisor 在 Flux 层组合。

4. Tokio：PollEvented → mio → park

Tokio 把 runtime 拆成 I/O driver、scheduler、timer¹。I/O 类型经 PollEvented 注册到 reactor；真正挂 fd 的是 Registration → IoDriver::add_source → mio::Registry::register⁸：

// tokio/src/io/poll_evented.rs L114–125（e77885a4）
pub(crate) fn new_with_interest_and_handle(
    mut io: E,
    interest: Interest,
    handle: scheduler::Handle,
) -> io::Result<Self> {
    let registration = Registration::new_with_interest_and_handle(&mut io, interest, handle)?;
    Ok(Self {
        io: Some(io),
        registration,
    })
}

// tokio/src/runtime/io/driver.rs L266–289（e77885a4）
pub(super) fn add_source(
    &self,
    source: &mut impl mio::event::Source,
    interest: Interest,
) -> io::Result<Arc<ScheduledIo>> {
    let scheduled_io = self.registrations.allocate(&mut self.synced.lock())?;
    let token = scheduled_io.token();
    self.registry.register(source, token, interest.to_mio())?;
    Ok(scheduled_io)
}

worker 无事可做时，IoDriver::park / park_timeout 进入 turn，核心是 self.poll.poll(events, max_wait)——Linux 上即 mio 封装的 epoll_wait：

// tokio/src/runtime/io/driver.rs L159–198（e77885a4）
pub(crate) fn park(&mut self, rt_handle: &driver::Handle) {
    let handle = rt_handle.io();
    self.turn(handle, None);
}

pub(crate) fn park_timeout(&mut self, rt_handle: &driver::Handle, duration: Duration) {
    let handle = rt_handle.io();
    self.turn(handle, Some(duration));
}

fn turn(&mut self, handle: &Handle, max_wait: Option<Duration>) {
    // ...
    match self.poll.poll(events, max_wait) {
        Ok(()) => {}
        Err(ref e) if e.kind() == io::ErrorKind::Interrupted => {}
        Err(e) => panic!("unexpected error when polling the I/O driver: {e:?}"),
    }
    // ... 按 token 分发就绪事件，唤醒对应 task ...
}

和 Reactor 的 subscribe / request 不同，Tokio 用 Waker 把 task 重新入队——语义不同，但「等 I/O → epoll 叫醒 → 继续推进」一致。入门见 async/await 与 Tokio⁹；I/O 驱动细节见 Tokio I/O driver³。

5. 「响应式」三层含义与对照

日常「响应式」至少三层，混用会吵不清楚：

Tokio 算不算「响应式」？

• Reactive Systems：算——事件驱动、异步消息、非阻塞 I/O。
• Reactive Streams：不算——没有 request(n) 协议；bounded mpsc 满时 send 等待，文档也写 channel provides backpressure¹³，那是 API 级流控，不是标准背压协议。
• 口语「异步非阻塞」：算——理念近，规范不等价。

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

把 Spring AI 的 Flux 说成「WebFlux 流式」容易误导——更准确是 Reactive Streams 流式；HTTP 层碰巧用了 Reactor Netty。

6. Spring AI：StreamAdvisor 双链

Spring AI 2.x 把 call 与 stream 拆成两条链：DefaultAroundAdvisorChain 里有 callAdvisors / streamAdvisors 两个 Deque；.call() → nextCall()，.stream() → nextStream()，注册可合并、执行不串台¹⁴¹⁵。

StreamAdvisor 约定返回 Flux<ChatClientResponse>¹⁶：

// spring-ai-client-chat/.../StreamAdvisor.java（d71200317）
public interface StreamAdvisor extends Advisor {
    Flux<ChatClientResponse> adviseStream(
        ChatClientRequest chatClientRequest,
        StreamAdvisorChain streamAdvisorChain);
}

链尾常见实现是 ChatModelStreamAdvisor：直接 chatModel.stream(prompt)，再 map 成 ChatClientResponse，并 publishOn(Schedulers.boundedElastic())：

// .../ChatModelStreamAdvisor.java L48–58（d71200317）
@Override
public Flux<ChatClientResponse> adviseStream(ChatClientRequest chatClientRequest,
        StreamAdvisorChain streamAdvisorChain) {
    Assert.notNull(chatClientRequest, "the chatClientRequest cannot be null");

    return this.chatModel.stream(chatClientRequest.prompt())
        .map(chatResponse -> ChatClientResponse.builder()
            .chatResponse(chatResponse)
            .context(Map.copyOf(chatClientRequest.context()))
            .build())
        .publishOn(Schedulers.boundedElastic());
}

DefaultChatClient 的 stream 规格里，终止于这条链：

// .../DefaultChatClient.java L724–729（d71200317）
Flux<ChatClientResponse> chatClientResponse = this.advisorChain.nextStream(chatClientRequest)
    .doOnError(observation::error)
    .doFinally(s -> observation.stop())
    .contextWrite(ctx -> ctx.put(ObservationThreadLocalAccessor.KEY, observation));

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，过滤 [DONE]，再 map 成 ChatCompletionChunk；tool-calling 片段用 windowUntil + reduce 合并¹⁷：

// models/spring-ai-deepseek/.../DeepSeekApi.java L168–212（d71200317）
return this.webClient.post()
    .uri(this.getEndpoint(chatRequest))
    .headers(headers -> headers.addAll(HttpHeaders.readOnlyHttpHeaders(additionalHttpHeader)))
    .body(Mono.just(chatRequest), ChatCompletionRequest.class)
    .retrieve()
    .bodyToFlux(String.class)
    .takeUntil(SSE_DONE_PREDICATE)
    .filter(SSE_DONE_PREDICATE.negate())
    .mapNotNull(content -> jsonHelper.fromJson(content, ChatCompletionChunk.class))
    .map(chunk -> {
        if (this.chunkMerger.isStreamingToolFunctionCall(chunk)) {
            isInsideTool.set(true);
        }
        return chunk;
    })
    .windowUntil(chunk -> {
        if (isInsideTool.get() && this.chunkMerger.isStreamingToolFunctionCallFinish(chunk)) {
            isInsideTool.set(false);
            return true;
        }
        return !isInsideTool.get();
    })
    .concatMapIterable(window -> {
        Mono<ChatCompletionChunk> monoChunk = window.reduce(this.chunkMerger::merge);
        return List.of(monoChunk);
    })
    .flatMap(mono -> mono);

可以看到：每个元素多半是 模型回复的 token 片段，不是把用户请求拆块发送；windowUntil 只在 tool-call 流式场景把碎片合成完整 chunk。

与 订票 Demo 对照：Demo 走 .call()；流式场景见 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. 逐块消费与聚合

打字机 UI 可以直接 subscribe 每个 ChatResponse chunk；Tool Calling 往往要等完整 AssistantMessage。MessageAggregator 在并行路径上把流式消息拼回去¹⁹：

// spring-ai-model/.../MessageAggregator.java（d71200317）
/**
 * 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) {
    AtomicReference<StringBuilder> messageTextContentRef = new AtomicReference<>(new StringBuilder());
    // ... thoughts / toolCalls / usage 等 AtomicReference ...
    return fluxChatResponse.doOnSubscribe(subscription -> {
        // 重置缓冲
    }) /* doOnNext 追加文本，doOnComplete 回调聚合结果 */;
}

DefaultAroundAdvisorChain.nextStream 还会经 ChatClientMessageAggregator 做链级聚合¹⁴。业务可选：

• 逐块：订阅 Flux，边到边渲染；
• 聚合后：在 Advisor after 或 MessageAggregator 回调里拿完整消息再调工具。

小结

• Reactor：流库 + Reactive Streams 背压；默认 subscribe 常是无界 request；网卡在 Netty。
• Tokio：捆绑 runtime；PollEvented → mio register → park 里 poll.poll（epoll）。
• 共性：Linux 上汇到 epoll / hrtimer / futex，不是「两套内核」。
• Spring AI：StreamAdvisor 返回 Flux；DeepSeek 路径是 WebClient.bodyToFlux，不必上 WebFlux Controller。

后续若继续写，可以单独拆 Reactor Netty EpollEventLoop 与 Tokio IoDriver 的线程模型对照。

References

• 上一篇 Spring AI 2 实战：订票聊天 Demo 的架构、ReAct 与 ToolCallingAdvisor 链
• 下一篇 Rust 生命周期标注在解决什么：返回值依赖、min 约束与 Java 为何不用