企业官网建设流程全解析

TanStack Query streamedQuery 完全指南：用 AsyncIterable 流式填充查询数据

【免费下载链接】query🤖 Powerful asynchronous state management, server-state utilities and data fetching for the web. TS/JS, React Query, Solid Query, Svelte Query and Vue Query.项目地址: https://gitcode.com/GitHub_Trending/qu/query

streamedQuery是 TanStack Query（本仓库 query-core）提供的实验性辅助函数，它把一个返回AsyncIterable的streamFn包装成标准queryFn，让查询数据可以像打字机一样逐块（chunk）写入缓存并即时渲染。本文结合仓库源码与测试用例，讲解它的状态生命周期、全部配置参数、refetch 三种模式的行为差异以及中止机制，帮助你在聊天、流式补全、长列表增量加载等场景中直接落地。

streamedQuery 是什么

在 streamedQuery 参考文档 中，官方对它的定位非常清晰：它是一个“helper function”，用于创建从一个 AsyncIterable 流式读取数据的查询函数。其行为可以归纳为三点：

• 最终数据是收到的所有 chunk 组成的数组（Array<TData>）；
• 查询在收到第一个 chunk 之前处于pending状态，收到之后立即转为success；
• 查询的fetchStatus会一直保持fetching，直到流结束。

这意味着你可以在数据尚未完全到达时就开始渲染“已到达的部分”，例如聊天机器人逐字输出回答、AI 流式补全、服务端分批推送的日志等场景，而不必等待整段数据返回。

在仓库中，它的实现位于 packages/query-core/src/streamedQuery.ts，由 query-core 以experimental_streamedQuery的名义导出（见 packages/query-core/src/index.ts#L44），React Query 通过export * from '@tanstack/query-core'将其直接暴露给使用方（见 packages/react-query/src/index.ts#L4）。

快速上手

官方文档给出的最小用法如下（注意当前的导入名是experimental_streamedQuery，文档中通常将其重命名为streamedQuery使用）：

import { experimental_streamedQuery as streamedQuery } from '@tanstack/react-query' const query = queryOptions({ queryKey: ['data'], queryFn: streamedQuery({ streamFn: fetchDataInChunks, }), })

其中fetchDataInChunks需要返回一个AsyncIterable（例如异步生成器），核心约束是每次yield一个 chunk：

async function* fetchDataInChunks() { const response = await fetch('/api/stream') const reader = response.body!.getReader() const decoder = new TextDecoder() while (true) { const { done, value } = await reader.read() if (done) break yield decoder.decode(value) } }

当流开始产出后，useQuery(query)拿到的data就是所有已产出 chunk 的数组；配合isFetching可以判断“流是否还在进行中”（详见下文“实战：Chat 示例”）。

Options 参数详解

streamedQuery接受一个参数对象，共四个字段，与 streamedQuery.ts 的类型定义 一一对应。

streamFn（必填）

• 签名：(context: QueryFunctionContext) => AsyncIterable<TQueryFnData> | Promise<AsyncIterable<TQueryFnData>>
• 必填。返回一个可异步迭代对象（AsyncIterable），负责产出要流式写入的数据块。
• 它接收标准的 QueryFunctionContext，因此可以拿到queryKey、client、meta等字段。

值得注意的是，源码中传给streamFn的 context 并非原样透传：它经过addConsumeAwareSignal包装，signal被定义为一个懒加载的 getter（详见 packages/query-core/src/utils.ts#L482-L510）。也就是说，只有当你真正读取context.signal时，中止信号才会被“消费”，从而决定是否在 refetch / 取消订阅时打断当前流（详见下文“取消与中止”）。

refetchMode（可选，默认'reset'）

• 取值：'append' | 'reset' | 'replace'
• 定义重新拉取（refetch）时如何处理旧数据。
• 默认值'reset'：refetch 时清空全部数据，查询回到pending状态；
• 'append'：新流产出的 chunk追加到已有数据之后；
• 'replace'：refetch 期间保留旧数据，等整个新流结束后，把新数据一次性写入缓存（整体替换）。

三种模式的差异在 streamedQuery 测试 中有非常直观的体现，下文“refetch 三种模式的行为对比”一节会逐一展开。

reducer（可选）

• 签名：(accumulator: TData, chunk: TQueryFnData) => TData
• 用于把流式 chunk（TQueryFnData）归约成最终的数据形态（TData）。
• 默认行为：当TData是数组时，把每个 chunk追加到数组末尾。其实现就是 utils.ts 中的addToEnd：[...items, item]，并支持可选的max上限（超出时从头部丢弃，streamedQuery 默认不启用该上限）。
• 如果TData不是数组，则必须提供自定义reducer（类型层面由SimpleStreamedQueryParams/ReducibleStreamedQueryParams联合类型强制约束，见 streamedQuery.ts#L16-L35）。

例如把一组 chunk 归约成一个对象：

streamedQuery({ streamFn: fetchNumbers, reducer: (acc, chunk) => ({ ...acc, [chunk]: true }), initialValue: {} as Record<number, boolean>, })

initialValue（可选）

• 类型：TData（当TData为数组时即为TQueryFnData[]）
• 默认值：空数组[]。
• 作用有二：一是第一个 chunk 到达之前作为占位数据；二是当流一个值都没有产出时，它作为最终结果返回。
• 当提供了自定义reducer时，initialValue为必填。

对应到源码，initialValue同时承担了“累积器起点”的角色——每个 chunk 写入缓存时，若缓存中尚无数据，会先以initialValue作为prev再执行 reducer（见 streamedQuery.ts#L107-L109）。

状态机与生命周期

streamedQuery最核心的使用要点是理解它的状态流转。参考文档指出：查询在收到第一个 chunk 前处于pending，之后转为success，而fetchStatus在流结束前一直保持fetching。

这个行为在 streamedQuery.test.tsx 的首个用例 中被精确验证。测试用了一个每 50ms 产出 1 个数字的异步生成器（共 3 个），对QueryObserver的结果断言如下：

这条时间线就是streamedQuery的“心电图”：第一个 chunk 决定status何时变为success，最后一个 chunk 决定fetchStatus何时从fetching变为idle。UI 上可以据此实现“先展示已到达内容 + 光标/占位动画，流结束后收起 loading 态”的典型流式体验。

此外，空流场景也有专门用例覆盖（streamedQuery.test.tsx#L131-L158）：一个立即结束、不产出任何值的异步生成器，会让查询直接从pending + fetching跳到success + idle，且data为默认的[]（即initialValue）。

源码实现原理

理解了状态流转后，再来看 streamedQuery.ts 的实现细节，你会发现它其实是一段相当精巧的“缓存写入循环”。

return async (context) => { const query = context.client .getQueryCache() .find({ queryKey: context.queryKey, exact: true }) const isRefetch = !!query && query.isFetched() if (isRefetch && refetchMode === 'reset') { query.setState({ ...query.resetState, fetchStatus: 'fetching' }) } // ... const stream = await streamFn(streamFnContext) const isReplaceRefetch = isRefetch && refetchMode === 'replace' for await (const chunk of stream) { if (cancelled) break if (isReplaceRefetch) { result = reducer(result, chunk) } else { context.client.setQueryData<TData>(context.queryKey, (prev) => reducer(prev === undefined ? initialValue : prev, chunk), ) } } if (isReplaceRefetch && !cancelled) { context.client.setQueryData<TData>(context.queryKey, result) } return context.client.getQueryData(context.queryKey) ?? initialValue }

几个关键设计值得注意：

1. refetch 判定：通过getQueryCache().find(...)找到当前 query，并用query.isFetched()判断是否为“重新拉取”。只有 refetch 才会触发reset/append/replace的分支逻辑。
2. 默认模式（reset）下逐 chunk 写缓存：每收到一个 chunk，就调用setQueryData并基于prev执行 reducer，因此观察者能立刻拿到增量数据；同时resetrefetch 会先把查询状态重置回pending（源码中的query.setState({ ...query.resetState, fetchStatus: 'fetching' })正是文档所说“erase all data and go back into pending state”的实现）。
3. replace模式延迟写回：refetch 期间把新 chunk 累积在局部变量result中，不触碰缓存（所以旧数据一直可见），待流结束后一次性setQueryData整体替换，从而避免“数据闪烁”。测试 streamedQuery.test.tsx#L269-L325 验证了这一点：refetch 过程中data仍是旧的[0, 1]，流结束后才变成新值[100, 101]。
4. 空流与最终返回值：函数末尾返回getQueryData(...) ?? initialValue，保证空流时initialValue成为最终数据。

另外，测试 “should not call reducer twice when refetchMode is replace” 还验证了 replace 模式下 reducer 不会重复执行：首次流式产出[1,2,3]，refetch 再次产出[1,2,3]，累计调用 6 次、但缓存数据始终是完整的[1,2,3]。

refetch 三种模式的行为对比

参考文档对refetchMode只给了三句话的说明，仓库测试则把每种模式的完整状态轨迹都画了出来，这里汇总成便于对照的表：

reset（默认）

见测试 streamedQuery.test.tsx#L160-L212：首次流结束后data为[0, 1]；调用refetch()后立即回到pending + fetching，data变为undefined；新流产出后逐步变为success + fetching并重新累积[0, 1]。旧数据被清空，界面需重新等待首个 chunk。

append

见测试 streamedQuery.test.tsx#L214-L267：refetch 时status保持success不变（data仍为[0, 1]），只是fetchStatus重新变回fetching；随后新 chunk 被追加，最终data变成[0, 1, 0, 1]。适合“加载更多 / 分页追加”类场景。

replace

见测试 streamedQuery.test.tsx#L269-L325：refetch 期间status保持success、data保持旧值[0, 1]，fetchStatus为fetching；整个流结束后一次性写入新数据[100, 101]。适合“静默刷新、整体换新”的场景，全程无数据闪断。

三者的共同点是：流未结束时fetchStatus都处于fetching，UI 都可以据此展示“进行中”的视觉反馈。

取消与中止

streamedQuery的中止行为有一个容易被忽略的关键点：是否中止取决于你的streamFn是否“消费”了context.signal。

回顾addConsumeAwareSignal（utils.ts#L482-L510）：signal是一个带记忆的 getter，第一次被读取时才真正取到AbortSignal并注册abort监听；一旦监听触发，会把streamedQuery内部的cancelled标志置为true，随后for await循环在下一个 chunk 处break，停止写缓存。

• 消费了 signal（例如把context.signal传给fetch）：refetch 或取消订阅时会中止当前流。测试 “should abort ongoing stream when refetch happens” 与 “should abort when unsubscribed” 验证了这一点：refetch 后旧流不再继续产出，取消订阅后新 chunk 也不会再写入。
• 没有消费 signal：流会继续跑完。测试 “should not abort when signal not consumed” 证明，即使已经取消订阅，后续 chunk 仍会继续写入缓存。

换句话说：如果你需要“组件卸载即停流”的能力，务必在streamFn内部读取并使用context.signal。

实战：Chat 示例

官方文档推荐通过 examples/react/chat 示例 观察streamedQuery的实际效果。这是一个最小可运行的打字机式聊天应用，运行方式见 examples/react/chat/README.md（npm install后npm run dev）。

其核心在 examples/react/chat/src/chat.ts：chatAnswer返回一个手写的异步迭代器对象，每 100~400ms 随机产出答案中的一个单词：

function chatAnswer(_question: string) { return { async *[Symbol.asyncIterator]() { const answer = answers[Math.floor(Math.random() * answers.length)] let index = 0 while (index < answer.length) { await new Promise((resolve) => setTimeout(resolve, 100 + Math.random() * 300), ) yield answer[index++] } }, } } export const chatQueryOptions = (question: string) => queryOptions({ queryKey: ['chat', question], queryFn: streamedQuery({ streamFn: () => chatAnswer(question), }), staleTime: Infinity, })

消费侧在 examples/react/chat/src/index.tsx：每个问题对应一个queryKey为['chat', question]的查询，data是单词数组，用data.join(' ')渲染成句子；isFetching为true时给消息气泡附加inProgress标记，形成“正在打字”的效果：

function ChatMessage({ question }: { question: string }) { const { error, data = [], isFetching } = useQuery(chatQueryOptions(question)) if (error) return 'An error has occurred: ' + error.message return ( <div> <Message message={{ content: question, isQuestion: true }} /> <Message inProgress={isFetching} message={{ content: data.join(' '), isQuestion: false }} /> </div> ) }

这个示例恰好演示了streamedQuery的典型配方：每个独立请求使用独立queryKey+staleTime: Infinity避免自动重新拉取，配合useQuery的data/isFetching驱动逐字渲染。

错误处理与边界情况

streamedQuery同样遵循 TanStack Query 的错误模型：streamFn或迭代过程中抛出的错误会被查询捕获并进入error状态。测试 “should keep error state on reset refetch when initialData is defined” 和 “should treat a fetch after an initial error as a refetch for reset mode” 验证了两个细节：

• 使用initialData时，reset模式的 refetch 失败后，data会回退到initialData对应的累积结果，error被保留；
• 首次拉取失败后再 refetch，会被当作一次resetrefetch 处理，查询回到pending + fetching并清除error。

另外，流中的 chunk 本身也可以是数组（即TQueryFnData为Array），测试 “should allow Arrays to be returned from the stream” 验证了data会变成数组的数组（如[[0, 0], [1, 1]]），此时若想展平数据，就需要自定义reducer。

实验性状态与适用范围

最后需要提醒：streamedQuery目前被标记为experimental（导出名为experimental_streamedQuery），项目团队之所以保留experimental前缀，是为了收集社区反馈后再稳定 API 形态。参考文档也明确说明：如果你试用过该 API 并有反馈，可以提交给官方讨论区。因此在生产环境中使用时，建议：

• 关注 query-core 的 CHANGELOG 与 react-query 的 CHANGELOG，留意streamedQuery从实验性转正或 API 调整的公告；
• 由于它是 query-core 层的通用能力，除了 React，Preact Query（preact-query 测试）与 Vue Query（vue-query 测试）同样可以直接使用experimental_streamedQuery；
• 与useSuspenseQuery组合时，Suspense 会在收到第一个 chunk 后释放（见 react-query 的 useSuspenseQuery 测试），适合构建“先占位、后流式填充”的体验。

总而言之，streamedQuery的价值在于把“服务端持续推送、客户端逐块消费”这一异步模式无缝接入 TanStack Query 的缓存与状态体系：状态机清晰、refetch 策略可配置、取消语义完整。参考本文的状态时间线与源码行为，你可以在自己的项目中复刻 Chat 示例，也可以用它支撑任何基于 AsyncIterable 的增量数据渲染需求。

【免费下载链接】query🤖 Powerful asynchronous state management, server-state utilities and data fetching for the web. TS/JS, React Query, Solid Query, Svelte Query and Vue Query.项目地址: https://gitcode.com/GitHub_Trending/qu/query