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

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/querystreamedQuery是 TanStack Query本仓库 query-core提供的实验性辅助函数它把一个返回AsyncIterable的streamFn包装成标准queryFn让查询数据可以像打字机一样逐块chunk写入缓存并即时渲染。本文结合仓库源码与测试用例讲解它的状态生命周期、全部配置参数、refetch 三种模式的行为差异以及中止机制帮助你在聊天、流式补全、长列表增量加载等场景中直接落地。streamedQuery 是什么在 streamedQuery 参考文档 中官方对它的定位非常清晰它是一个“helper function”用于创建从一个 AsyncIterable 流式读取数据的查询函数。其行为可以归纳为三点最终数据是收到的所有 chunk 组成的数组ArrayTData查询在收到第一个 chunk 之前处于pending状态收到之后立即转为success查询的fetchStatus会一直保持fetching直到流结束。这意味着你可以在数据尚未完全到达时就开始渲染“已到达的部分”例如聊天机器人逐字输出回答、AI 流式补全、服务端分批推送的日志等场景而不必等待整段数据返回。在仓库中它的实现位于 packages/query-core/src/streamedQuery.ts由 query-core 以experimental_streamedQuery的名义导出见 packages/query-core/src/index.ts#L44React 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一个 chunkasync 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) AsyncIterableTQueryFnData | PromiseAsyncIterableTQueryFnData必填。返回一个可异步迭代对象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时如何处理旧数据。默认值resetrefetch 时清空全部数据查询回到pending状态append新流产出的 chunk追加到已有数据之后replacerefetch 期间保留旧数据等整个新流结束后把新数据一次性写入缓存整体替换。三种模式的差异在 streamedQuery 测试 中有非常直观的体现下文“refetch 三种模式的行为对比”一节会逐一展开。reducer可选签名(accumulator: TData, chunk: TQueryFnData) TData用于把流式 chunkTQueryFnData归约成最终的数据形态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 Recordnumber, 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的结果断言如下时间点statusfetchStatusdata订阅后立即pendingfetchingundefined50ms收到 chunk 0successfetching[0]100ms收到 chunk 1successfetching[0, 1]150ms流结束successidle[0, 1, 2]这条时间线就是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.setQueryDataTData(context.queryKey, (prev) reducer(prev undefined ? initialValue : prev, chunk), ) } } if (isReplaceRefetch !cancelled) { context.client.setQueryDataTData(context.queryKey, result) } return context.client.getQueryData(context.queryKey) ?? initialValue }几个关键设计值得注意refetch 判定通过getQueryCache().find(...)找到当前 query并用query.isFetched()判断是否为“重新拉取”。只有 refetch 才会触发reset/append/replace的分支逻辑。默认模式reset下逐 chunk 写缓存每收到一个 chunk就调用setQueryData并基于prev执行 reducer因此观察者能立刻拿到增量数据同时resetrefetch 会先把查询状态重置回pending源码中的query.setState({ ...query.resetState, fetchStatus: fetching })正是文档所说“erase all data and go back into pending state”的实现。replace模式延迟写回refetch 期间把新 chunk 累积在局部变量result中不触碰缓存所以旧数据一直可见待流结束后一次性setQueryData整体替换从而避免“数据闪烁”。测试 streamedQuery.test.tsx#L269-L325 验证了这一点refetch 过程中data仍是旧的[0, 1]流结束后才变成新值[100, 101]。空流与最终返回值函数末尾返回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 fetchingdata变为undefined新流产出后逐步变为success fetching并重新累积[0, 1]。旧数据被清空界面需重新等待首个 chunk。append见测试 streamedQuery.test.tsx#L214-L267refetch 时status保持success不变data仍为[0, 1]只是fetchStatus重新变回fetching随后新 chunk 被追加最终data变成[0, 1, 0, 1]。适合“加载更多 / 分页追加”类场景。replace见测试 streamedQuery.test.tsx#L269-L325refetch 期间status保持success、data保持旧值[0, 1]fetchStatus为fetching整个流结束后一次性写入新数据[100, 101]。适合“静默刷新、整体换新”的场景全程无数据闪断。三者的共同点是流未结束时fetchStatus都处于fetchingUI 都可以据此展示“进行中”的视觉反馈。取消与中止streamedQuery的中止行为有一个容易被忽略的关键点是否中止取决于你的streamFn是否“消费”了context.signal。回顾addConsumeAwareSignalutils.ts#L482-L510signal是一个带记忆的 getter第一次被读取时才真正取到AbortSignal并注册abort监听一旦监听触发会把streamedQuery内部的cancelled标志置为true随后for await循环在下一个 chunk 处break停止写缓存。消费了 signal例如把context.signal传给fetchrefetch 或取消订阅时会中止当前流。测试 “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.mdnpm install后npm run dev。其核心在 examples/react/chat/src/chat.tschatAnswer返回一个手写的异步迭代器对象每 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的典型配方每个独立请求使用独立queryKeystaleTime: 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 层的通用能力除了 ReactPreact Querypreact-query 测试与 Vue Queryvue-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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考