TanStack Start 服务端函数流式传输实战类型安全的 ReadableStream 与异步生成器【免费下载链接】router A client-first, server-capable, fully type-safe router and full-stack framework for the web (React and more).项目地址: https://gitcode.com/GitHub_Trending/ro/router随着 AI 应用的兴起从服务端向客户端流式传输数据已经成为 Web 开发中的高频需求——无论是大语言模型的逐 token 输出、实时日志推送还是长列表的渐进式加载。在 TanStack Start本项目中的 React Start 框架即tanstack/react-start中实现流式传输并不复杂而且最大的亮点在于流式传输的数据是类型安全的。本文基于 官方指南 展开结合仓库中的完整示例与底层源码讲解在 Server Function 中通过ReadableStream与异步生成器Async Generator两种方式实现端到端类型安全的流式数据传输读完即可直接落地到自己的 TanStack Start 项目中。一、背景为什么 Server Function 适合做流式传输TanStack Start 的 Server Function 是定义在服务端、却可以在应用任意位置loader、组件、hook 或其他 Server Function无缝调用的 RPC 函数。它天然适合作为流式数据的出口原因有三跨网络边界的类型安全Server Function 的输入与输出都会经过类型检查默认strict模式返回值必须是可序列化的而ReadableStream与异步生成器正是框架支持的特殊可序列化类型调用方式统一客户端调用 Server Function 与调用普通异步函数体验一致无需手动拼接 fetch 与解析协议渐进式呈现服务端可以边生成边推送客户端边接收边渲染显著改善长耗时任务的感知性能。仓库中的官方示例 start-streaming-data-from-server-functions 演示了这两种方式的完整实现其路由页面包含两个按钮分别触发ReadableStream与异步生成器版本的流式响应将 10 条随机数字消息逐条推送到页面。二、前置知识Server Function 的基础形态在进入流式传输之前先回顾 Server Function 的基本写法。它通过createServerFn()创建使用.handler()定义服务端逻辑import { createServerFn } from tanstack/react-start // GET 请求默认 export const getData createServerFn().handler(async () { return { message: Hello from server! } }) // POST 请求 export const saveData createServerFn({ method: POST }).handler(async () { return { success: true } })关于 Server Function 的完整用法validator 校验、错误处理、重定向、文件组织等可参考 服务端函数指南。流式传输的所有代码都建立在createServerFn().handler(...)这一基础之上——区别仅在于 handler 的返回值类型。三、方式一类型化的 ReadableStream3.1 服务端返回一个泛型化的ReadableStream在 Server Function 中直接返回一个ReadableStreamT框架会将每个 chunk 的类型信息一并传递给客户端。以下示例来自仓库官方示例的简化版本流式返回一个Message数组type Message { content: string } /** 该 Server Function 返回一个 ReadableStream 将 Message 类型的 chunk 流式推送给客户端。 */ const streamingResponseFn createServerFn().handler(async () { // 这是你想以 chunk 形式发送给客户端的消息数组 const messages: Message[] generateMessages() // 这个 ReadableStream 是类型化的 // 因此每个 chunk 都是 Message 类型。 const stream new ReadableStreamMessage({ async start(controller) { for (const message of messages) { // 发送消息 controller.enqueue(message) } controller.close() }, }) return stream })关键点解析new ReadableStreamMessage的泛型参数会沿着 Server Function 的返回类型一路传递到客户端客户端拿到的 stream 元素类型即为Messagecontroller.enqueue(message)每调用一次就向客户端推送一个 chunk所有消息推送完毕后必须调用controller.close()告知接收端流已结束实际业务中start()内部可以是异步的——例如逐条从数据库、队列或 LLM 响应中读取并enqueue完全符合流式语义。仓库的完整示例中chunk 类型模拟了 OpenAI 的流式响应格式包含choices[].delta.content的TextPart结构并使用zod在服务端解析与校验每个 chunk见 示例路由。3.2 客户端通过getReader()逐块消费客户端消费该流时streamed chunks 会保持完整的类型信息const [message, setMessage] useState() const getTypedReadableStreamResponse useCallback(async () { const response await streamingResponseFn() if (!response) { return } const reader response.getReader() let done false while (!done) { const { value, done: doneReading } await reader.read() done doneReading if (value) { // 注意这里的 value 是 Message | undefined // 因为它来自类型化的 ReadableStream const chunk value.content setMessage((prev) prev chunk) } } }, [])这段代码体现了标准 Streams API 的消费模式await streamingResponseFn()返回的就是一个ReadableStreamMessageresponse.getReader()拿到读取器循环中await reader.read()每次返回{ value, done }done为true表示流结束类型保障的关键点value.content之所以能通过类型检查正是因为ReadableStreamMessage的泛型信息跨越了客户端/服务端边界被保留了下来每收到一个 chunk 就通过setMessage追加到 UI 状态实现逐条渲染的渐进式体验。示例中将其挂载到按钮的onClick点击后 10 条消息以约 500ms 的间隔逐条出现在页面上见 示例路由。四、方式二Server Function 中的异步生成器4.1 服务端用 async generator 简化流式逻辑异步生成器是更简洁、可读性更强的等价方案——相同的流式结果代码却大幅精简const streamingWithAnAsyncGeneratorFn createServerFn().handler( async function* () { const messages: Message[] generateMessages() for (const msg of messages) { await sleep(500) // 流式传输的 chunk 仍然被类型化为 Message yield msg } }, )在这里handler 传入的是一个异步生成器函数async function* ()每次yield一个 chunkawait sleep(500)模拟真实的网络/生成延迟让客户端能看到逐条到达的效果生成器函数天然支持中途等待异步操作如等待数据库查询、LLM 的下一个 token无需手动管理controller的生命周期——流的开始、结束与错误传播都由生成器语义自动完成每个yield出去的值同样携带类型信息客户端拿到的仍然是Message。仓库示例中的generateMessages()生成 10 条符合TextPartschema 的消息sleep()辅助函数用于模拟每条消息之间的延迟见 示例路由。4.2 客户端用for await...of优雅消费客户端代码也随之变得更加精简const getResponseFromTheAsyncGenerator useCallback(async () { for await (const msg of await streamingWithAnAsyncGeneratorFn()) { const chunk msg.content setMessages((prev) prev chunk) } }, [])for await...of语法会自动处理读取器的获取、逐块迭代与流的收尾工作无需手动编写while循环与done判断。这是两种方式中最为推荐的写法——服务端用生成器表达逐步产出客户端用for await表达逐条消费语义完全对仗。五、两种方式对比与选型建议维度ReadableStream异步生成器Async Generator服务端代码需手动管理controllerenqueue / closeyield即推送框架管理生命周期客户端代码getReader()while循环for await...of更简洁类型安全✅ chunk 类型贯穿两端✅ 同样类型安全适用场景需要精细控制流如背压、自定义取消逻辑绝大多数逐步产出数据的场景推荐首选与 AI 流式输出的契合度可直接映射底层 LLM 流可用yield逐 token 转发 LLM 响应选型建议如果你只是想把一批数据逐条推给客户端AI 对话流、日志流、进度通知优先选择异步生成器——代码量最少、可读性最高如果需要对流本身做精细控制例如手动处理取消、错误恢复、或对接已有的ReadableStream数据源则使用ReadableStream方案。六、源码级原理流式数据是如何跨过网络边界的两种方式的类型安全与流式能力并非魔法而是由仓库中一套完整的序列化 帧协议基础设施支撑的。理解底层原理有助于排查问题与扩展能力。6.1 服务端识别流式返回值并做多路复用当 Server Function 的 handler 返回ReadableStream或异步生成器时服务端处理器会走一条专门的序列化链路见 服务端处理实现序列化开始时createRawStreamRPCPlugin插件定义于 RawStream.ts会拦截序列化过程中遇到的RawStream标记为其分配自增的streamId并登记对应的原始流若序列化同步完成且没有流则退回普通的 JSON 响应Content-Type: application/json一旦存在流或序列化未同步完成则启用帧协议framed protocolcreateMultiplexedStream见 frame-protocol.ts将 JSON 结果流与若干原始二进制流合成为一个ReadableStreamUint8Array逐帧编码JSON帧streamId 固定为 0承载序列化后的结果元数据CHUNK帧携带某个流非零 streamId的一段数据END帧标记某个流结束ERROR帧标记某个流出错响应头Content-Type设置为带版本号的帧协议类型TSS_CONTENT_TYPE_FRAMED_VERSIONED并附上X_TSS_SERIALIZED: true标记。对于异步生成器或返回PromiseReadableStream这类序列化过程中才陆续发现流的情况服务端还实现了**延迟流注册late stream registration**机制初始同步阶段之后新发现的流会写入一个TransformStream通道由pumpLateStreams持续转发避免竞态丢流见 server-functions-handler.ts。6.2 客户端帧解码与流重建客户端侧的serverFnFetcher见 serverFnFetcher.ts负责识别帧协议响应检测到帧协议后调用createFrameDecoder(response.body)实现见 frame-decoder.ts创建解码器解码器按帧头类型 streamId 长度解析出两条输出一条 NDJSON 格式的chunks: ReadableStreamstring与一个按 ID 取流的getStream(id)接口客户端侧的反序列化插件createRawStreamDeserializePlugin见 RawStream.ts在反序列化元数据时遇到RawStream引用就通过getStream(id)重建出对应的ReadableStreamUint8Array最终客户端拿到的ReadableStreamT与for await可迭代对象就是由这些重建的流包装而来——这也是类型信息得以贯穿两端的根本原因chunk 的类型定义在编译期通过ReadableStreamT与生成器返回类型传递运行时则由帧协议按流 ID 精确还原。值得留意的是帧解码器还内置了防滥用保护单帧载荷上限 16MiB、缓冲上限 32MiB、最多 1024 条流、最多 100000 帧防止恶意或异常响应导致内存/CPU 耗尽见 frame-decoder.ts。上述序列化插件 帧协议的往返行为在 RawStream 单元测试 中有完整的覆盖验证。七、把示例跑起来仓库提供了开箱即用的示例可以直接本地运行# 基于示例初始化一个全新项目示例位于 examples/react/start-streaming-data-from-server-functions npx gitpick TanStack/router/tree/main/examples/react/start-streaming-data-from-server-functions start-streaming-data-from-server-functions # 或直接在仓库示例目录中安装并启动 pnpm install pnpm dev生产构建同样简单pnpm build示例的依赖与脚本定义见 package.json核心依赖为tanstack/react-start提供createServerFn、tanstack/react-router提供createFileRoute与zod定义 chunk 校验 schemapnpm dev以 Vite 开发模式启动并支持热更新。启动后访问首页点击 Get 10 random numbers (ReadableStream) 或 Get 10 random numbers (Async Generator Function) 按钮即可观察两条独立的流分别以约 500ms 的间隔逐条渲染 10 条随机数字。八、实战注意事项确认返回值可被流式序列化ReadableStream与异步生成器是框架认可的流式返回值strict类型检查默认开启若自定义类型无法通过序列化检查可参考 Server Functions 指南 中的strict配置按需放宽但运行时仍需遵守序列化规则。不要在服务端提前把所有数据塞进内存流式传输的意义在于边生成边发送。若在start()或生成器开头一次性Array.from构造完所有数据再推送就退化成了分批发送的整包响应失去了流式降低首字节延迟的意义。正确处理流的结束与取消ReadableStream方案务必在所有数据入队后调用controller.close()客户端若提前放弃消费如组件卸载应调用reader.cancel()帧解码器与服务端会相应释放资源。错误会跨端传播服务端流中抛出异常会以ERROR帧传输客户端reader.read()或for await循环会将其作为异常抛出可按常规try/catch处理。公开 API 请使用 Server RoutesServer Function 是应用内部的同源 RPC受 CSRF 中间件保护若需要允许外部系统调用的流式端点应改用 Server Routes。总结在 TanStack Start 中从 Server Function 向客户端流式传输数据是一条被框架完整支持的路径ReadableStreamT提供精细控制与完整类型异步生成器提供最简洁的编写体验而底层的帧协议、延迟流注册与客户端帧解码器共同保证了类型信息与流语义在客户端/服务端边界上的无损传递。无论是接入 LLM 流式输出、实时进度推送还是构建渐进式加载的长列表你都可以从本文的两种模式中直接选用一种开始实现。【免费下载链接】router A client-first, server-capable, fully type-safe router and full-stack framework for the web (React and more).项目地址: https://gitcode.com/GitHub_Trending/ro/router创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 TanStack Start 服务端函数流式传输实战：类型安全的 ReadableStream 与异步生成器
TanStack Start 服务端函数流式传输实战类型安全的 ReadableStream 与异步生成器【免费下载链接】router A client-first, server-capable, fully type-safe router and full-stack framework for the web (React and more).项目地址: https://gitcode.com/GitHub_Trending/ro/router随着 AI 应用的兴起从服务端向客户端流式传输数据已经成为 Web 开发中的高频需求——无论是大语言模型的逐 token 输出、实时日志推送还是长列表的渐进式加载。在 TanStack Start本项目中的 React Start 框架即tanstack/react-start中实现流式传输并不复杂而且最大的亮点在于流式传输的数据是类型安全的。本文基于 官方指南 展开结合仓库中的完整示例与底层源码讲解在 Server Function 中通过ReadableStream与异步生成器Async Generator两种方式实现端到端类型安全的流式数据传输读完即可直接落地到自己的 TanStack Start 项目中。一、背景为什么 Server Function 适合做流式传输TanStack Start 的 Server Function 是定义在服务端、却可以在应用任意位置loader、组件、hook 或其他 Server Function无缝调用的 RPC 函数。它天然适合作为流式数据的出口原因有三跨网络边界的类型安全Server Function 的输入与输出都会经过类型检查默认strict模式返回值必须是可序列化的而ReadableStream与异步生成器正是框架支持的特殊可序列化类型调用方式统一客户端调用 Server Function 与调用普通异步函数体验一致无需手动拼接 fetch 与解析协议渐进式呈现服务端可以边生成边推送客户端边接收边渲染显著改善长耗时任务的感知性能。仓库中的官方示例 start-streaming-data-from-server-functions 演示了这两种方式的完整实现其路由页面包含两个按钮分别触发ReadableStream与异步生成器版本的流式响应将 10 条随机数字消息逐条推送到页面。二、前置知识Server Function 的基础形态在进入流式传输之前先回顾 Server Function 的基本写法。它通过createServerFn()创建使用.handler()定义服务端逻辑import { createServerFn } from tanstack/react-start // GET 请求默认 export const getData createServerFn().handler(async () { return { message: Hello from server! } }) // POST 请求 export const saveData createServerFn({ method: POST }).handler(async () { return { success: true } })关于 Server Function 的完整用法validator 校验、错误处理、重定向、文件组织等可参考 服务端函数指南。流式传输的所有代码都建立在createServerFn().handler(...)这一基础之上——区别仅在于 handler 的返回值类型。三、方式一类型化的 ReadableStream3.1 服务端返回一个泛型化的ReadableStream在 Server Function 中直接返回一个ReadableStreamT框架会将每个 chunk 的类型信息一并传递给客户端。以下示例来自仓库官方示例的简化版本流式返回一个Message数组type Message { content: string } /** 该 Server Function 返回一个 ReadableStream 将 Message 类型的 chunk 流式推送给客户端。 */ const streamingResponseFn createServerFn().handler(async () { // 这是你想以 chunk 形式发送给客户端的消息数组 const messages: Message[] generateMessages() // 这个 ReadableStream 是类型化的 // 因此每个 chunk 都是 Message 类型。 const stream new ReadableStreamMessage({ async start(controller) { for (const message of messages) { // 发送消息 controller.enqueue(message) } controller.close() }, }) return stream })关键点解析new ReadableStreamMessage的泛型参数会沿着 Server Function 的返回类型一路传递到客户端客户端拿到的 stream 元素类型即为Messagecontroller.enqueue(message)每调用一次就向客户端推送一个 chunk所有消息推送完毕后必须调用controller.close()告知接收端流已结束实际业务中start()内部可以是异步的——例如逐条从数据库、队列或 LLM 响应中读取并enqueue完全符合流式语义。仓库的完整示例中chunk 类型模拟了 OpenAI 的流式响应格式包含choices[].delta.content的TextPart结构并使用zod在服务端解析与校验每个 chunk见 示例路由。3.2 客户端通过getReader()逐块消费客户端消费该流时streamed chunks 会保持完整的类型信息const [message, setMessage] useState() const getTypedReadableStreamResponse useCallback(async () { const response await streamingResponseFn() if (!response) { return } const reader response.getReader() let done false while (!done) { const { value, done: doneReading } await reader.read() done doneReading if (value) { // 注意这里的 value 是 Message | undefined // 因为它来自类型化的 ReadableStream const chunk value.content setMessage((prev) prev chunk) } } }, [])这段代码体现了标准 Streams API 的消费模式await streamingResponseFn()返回的就是一个ReadableStreamMessageresponse.getReader()拿到读取器循环中await reader.read()每次返回{ value, done }done为true表示流结束类型保障的关键点value.content之所以能通过类型检查正是因为ReadableStreamMessage的泛型信息跨越了客户端/服务端边界被保留了下来每收到一个 chunk 就通过setMessage追加到 UI 状态实现逐条渲染的渐进式体验。示例中将其挂载到按钮的onClick点击后 10 条消息以约 500ms 的间隔逐条出现在页面上见 示例路由。四、方式二Server Function 中的异步生成器4.1 服务端用 async generator 简化流式逻辑异步生成器是更简洁、可读性更强的等价方案——相同的流式结果代码却大幅精简const streamingWithAnAsyncGeneratorFn createServerFn().handler( async function* () { const messages: Message[] generateMessages() for (const msg of messages) { await sleep(500) // 流式传输的 chunk 仍然被类型化为 Message yield msg } }, )在这里handler 传入的是一个异步生成器函数async function* ()每次yield一个 chunkawait sleep(500)模拟真实的网络/生成延迟让客户端能看到逐条到达的效果生成器函数天然支持中途等待异步操作如等待数据库查询、LLM 的下一个 token无需手动管理controller的生命周期——流的开始、结束与错误传播都由生成器语义自动完成每个yield出去的值同样携带类型信息客户端拿到的仍然是Message。仓库示例中的generateMessages()生成 10 条符合TextPartschema 的消息sleep()辅助函数用于模拟每条消息之间的延迟见 示例路由。4.2 客户端用for await...of优雅消费客户端代码也随之变得更加精简const getResponseFromTheAsyncGenerator useCallback(async () { for await (const msg of await streamingWithAnAsyncGeneratorFn()) { const chunk msg.content setMessages((prev) prev chunk) } }, [])for await...of语法会自动处理读取器的获取、逐块迭代与流的收尾工作无需手动编写while循环与done判断。这是两种方式中最为推荐的写法——服务端用生成器表达逐步产出客户端用for await表达逐条消费语义完全对仗。五、两种方式对比与选型建议维度ReadableStream异步生成器Async Generator服务端代码需手动管理controllerenqueue / closeyield即推送框架管理生命周期客户端代码getReader()while循环for await...of更简洁类型安全✅ chunk 类型贯穿两端✅ 同样类型安全适用场景需要精细控制流如背压、自定义取消逻辑绝大多数逐步产出数据的场景推荐首选与 AI 流式输出的契合度可直接映射底层 LLM 流可用yield逐 token 转发 LLM 响应选型建议如果你只是想把一批数据逐条推给客户端AI 对话流、日志流、进度通知优先选择异步生成器——代码量最少、可读性最高如果需要对流本身做精细控制例如手动处理取消、错误恢复、或对接已有的ReadableStream数据源则使用ReadableStream方案。六、源码级原理流式数据是如何跨过网络边界的两种方式的类型安全与流式能力并非魔法而是由仓库中一套完整的序列化 帧协议基础设施支撑的。理解底层原理有助于排查问题与扩展能力。6.1 服务端识别流式返回值并做多路复用当 Server Function 的 handler 返回ReadableStream或异步生成器时服务端处理器会走一条专门的序列化链路见 服务端处理实现序列化开始时createRawStreamRPCPlugin插件定义于 RawStream.ts会拦截序列化过程中遇到的RawStream标记为其分配自增的streamId并登记对应的原始流若序列化同步完成且没有流则退回普通的 JSON 响应Content-Type: application/json一旦存在流或序列化未同步完成则启用帧协议framed protocolcreateMultiplexedStream见 frame-protocol.ts将 JSON 结果流与若干原始二进制流合成为一个ReadableStreamUint8Array逐帧编码JSON帧streamId 固定为 0承载序列化后的结果元数据CHUNK帧携带某个流非零 streamId的一段数据END帧标记某个流结束ERROR帧标记某个流出错响应头Content-Type设置为带版本号的帧协议类型TSS_CONTENT_TYPE_FRAMED_VERSIONED并附上X_TSS_SERIALIZED: true标记。对于异步生成器或返回PromiseReadableStream这类序列化过程中才陆续发现流的情况服务端还实现了**延迟流注册late stream registration**机制初始同步阶段之后新发现的流会写入一个TransformStream通道由pumpLateStreams持续转发避免竞态丢流见 server-functions-handler.ts。6.2 客户端帧解码与流重建客户端侧的serverFnFetcher见 serverFnFetcher.ts负责识别帧协议响应检测到帧协议后调用createFrameDecoder(response.body)实现见 frame-decoder.ts创建解码器解码器按帧头类型 streamId 长度解析出两条输出一条 NDJSON 格式的chunks: ReadableStreamstring与一个按 ID 取流的getStream(id)接口客户端侧的反序列化插件createRawStreamDeserializePlugin见 RawStream.ts在反序列化元数据时遇到RawStream引用就通过getStream(id)重建出对应的ReadableStreamUint8Array最终客户端拿到的ReadableStreamT与for await可迭代对象就是由这些重建的流包装而来——这也是类型信息得以贯穿两端的根本原因chunk 的类型定义在编译期通过ReadableStreamT与生成器返回类型传递运行时则由帧协议按流 ID 精确还原。值得留意的是帧解码器还内置了防滥用保护单帧载荷上限 16MiB、缓冲上限 32MiB、最多 1024 条流、最多 100000 帧防止恶意或异常响应导致内存/CPU 耗尽见 frame-decoder.ts。上述序列化插件 帧协议的往返行为在 RawStream 单元测试 中有完整的覆盖验证。七、把示例跑起来仓库提供了开箱即用的示例可以直接本地运行# 基于示例初始化一个全新项目示例位于 examples/react/start-streaming-data-from-server-functions npx gitpick TanStack/router/tree/main/examples/react/start-streaming-data-from-server-functions start-streaming-data-from-server-functions # 或直接在仓库示例目录中安装并启动 pnpm install pnpm dev生产构建同样简单pnpm build示例的依赖与脚本定义见 package.json核心依赖为tanstack/react-start提供createServerFn、tanstack/react-router提供createFileRoute与zod定义 chunk 校验 schemapnpm dev以 Vite 开发模式启动并支持热更新。启动后访问首页点击 Get 10 random numbers (ReadableStream) 或 Get 10 random numbers (Async Generator Function) 按钮即可观察两条独立的流分别以约 500ms 的间隔逐条渲染 10 条随机数字。八、实战注意事项确认返回值可被流式序列化ReadableStream与异步生成器是框架认可的流式返回值strict类型检查默认开启若自定义类型无法通过序列化检查可参考 Server Functions 指南 中的strict配置按需放宽但运行时仍需遵守序列化规则。不要在服务端提前把所有数据塞进内存流式传输的意义在于边生成边发送。若在start()或生成器开头一次性Array.from构造完所有数据再推送就退化成了分批发送的整包响应失去了流式降低首字节延迟的意义。正确处理流的结束与取消ReadableStream方案务必在所有数据入队后调用controller.close()客户端若提前放弃消费如组件卸载应调用reader.cancel()帧解码器与服务端会相应释放资源。错误会跨端传播服务端流中抛出异常会以ERROR帧传输客户端reader.read()或for await循环会将其作为异常抛出可按常规try/catch处理。公开 API 请使用 Server RoutesServer Function 是应用内部的同源 RPC受 CSRF 中间件保护若需要允许外部系统调用的流式端点应改用 Server Routes。总结在 TanStack Start 中从 Server Function 向客户端流式传输数据是一条被框架完整支持的路径ReadableStreamT提供精细控制与完整类型异步生成器提供最简洁的编写体验而底层的帧协议、延迟流注册与客户端帧解码器共同保证了类型信息与流语义在客户端/服务端边界上的无损传递。无论是接入 LLM 流式输出、实时进度推送还是构建渐进式加载的长列表你都可以从本文的两种模式中直接选用一种开始实现。【免费下载链接】router A client-first, server-capable, fully type-safe router and full-stack framework for the web (React and more).项目地址: https://gitcode.com/GitHub_Trending/ro/router创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考