响应式事件架构实战：用 RxJS 终结前端回调地狱

不知道从什么时候开始rea这个词在我这边从一个随手敲下的缩写变成了一个内部项目的代号。当时我们遇到一个最典型的回调地狱场景页面上的交互逻辑越堆越厚搜索联想、拖拽排序、无限滚动、消息推送全部搅在一起事件源多到根本理不清触发顺序代码里到处是防抖函数、竞态处理标志位和清理定时器的散弹式修补。后来我把这套问题的解法收敛成一个名字Reactive Event Architecture简称 rea也就是用响应式事件的思路重新组织前端的一切异步交互。这篇文章我会把 rea 从需求起点到落地踩坑的完整过程拆开讲包含核心原理解析、选型取舍、高频场景的代码实现以及我们后来踩到的四个深坑和对应的排查链路。适合正在被复杂交互逻辑折磨的前端开发也适合想理解响应式编程到底解决了什么问题、而不只是会调几个 API 的同学。1. 一切源于回调地狱rea项目的真实起点1.1 当时的代码到底乱成了什么样先说场景。一个管理后台的工单列表页功能不算复杂但很典型顶部一个搜索输入框带 500ms 防抖联想列表需要滚动到底部时自动加载下一页每一行工单可以拖拽调整优先级同时页面还要接收 WebSocket 推送的新工单提醒实时刷新列表。如果按照传统命令式写法这页面的状态变量大概会是inputValue、searchResult、pageNum、totalCount、listData、loading、dragOrder、socketMessage……每个变量都会被多个函数读写而每次读写都要考虑现在用户正在做什么、异步请求返回了没有、旧请求要不要丢弃。我见过最真实的代码是这样的let timer null; let requestSeq 0; function handleSearchInput(value) { clearTimeout(timer); timer setTimeout(() { const seq requestSeq; fetchSuggestions(value).then((res) { if (seq ! requestSeq) return; // 防止旧请求覆盖新结果 renderSuggestions(res.data); }); }, 500); }这个写法本身没毛病但它只是整个页面的一个角落。当搜索、拖拽、分页、推送四个功能叠加在一起每一个异步源都要做类似的清理 序号 判断操作代码根本没法维护。我们当时的新人接手这个页面光理清所有分支就花了一周改一个 bug 的平均耗时是半天。1.2 从问题反推需求rea要解决的四件事当时团队复盘列出的痛点后来直接变成了 rea 的需求清单第一事件源太多触发顺序不可控。用户的输入、滚动、拖拽、服务端推送这些都是独立发生的异步事件但它们最终都要更新同一个页面状态。命令式代码里没有事件的先后关系这个概念只有函数谁调用了谁所以一旦时序出问题bug 极难复现。第二每个异步操作都要手动做竞态防护。上面代码里的requestSeq就是为了防止上一次搜索请求比下一次慢导致结果错乱。这种防护代码写多了以后业务逻辑里超过一半的代码其实都在处理时序问题而不是真正的业务逻辑。第三数据流的变换和组合缺少统一语法。搜索输入要防抖、过滤、映射为请求参数推送消息要过滤类型、映射为新的工单对象、跟现有列表做合并去重。这些操作在传统写法里散落在各函数里没有一个统一的地方表达数据从源头到 UI 的完整走向。第四很难扩展新的事件源。比如产品说在用户停止输入 3 秒后自动聚焦到第一条联想结果传统写法又得在某个函数里硬塞加一段逻辑牵一发而动全身。所以 rea 的核心定位不是发明新框架而是把页面的所有交互全部显式建模为事件流让状态变更的链路清晰可推理。2. 响应式事件架构的核心原型从观察者模式到数据流管道2.1 响应式不是玄学它只是把事件升级成了流理解 rea 甚至整个响应式编程第一步要把思维从事件是一次性的切换成事件是一根持续流动的水管。普通的事件监听好比你站在路边招手——每次来一辆车事件司机回调函数踩一脚刹车处理一下。而流更像是自来水管——水数据持续不断地流过来你在水管中间装各种过滤器、分流器、蓄水池每经过一个装置水就变换一次形态。这个思想落到代码层面就是三个核心概念Observable可观察对象代表一段随时间产生数据的事件源。它不是一次性返回结果而是在未来持续发数据。Observer观察者一个对象包含next、error、complete三个方法分别处理流中来的数据、错误、以及流结束。Subscription订阅关系调用observable.subscribe(observer)后返回的连接线可以用来取消监听这是解决内存泄漏和停止接收事件的关键。举个最简单的例子把一个按钮点击事件变成流import { fromEvent } from rxjs; const clickStream fromEvent(document.getElementById(saveBtn), click); const subscription clickStream.subscribe({ next: (event) { console.log(点击保存按钮当前坐标, event.clientX, event.clientY); }, error: (err) console.error(事件流异常, err), complete: () console.log(事件流结束), }); // 不再需要监听时调用 unsubscribe 释放 // subscription.unsubscribe();这里fromEvent返回的就是一个 Observable。订阅后每次点击都会走一遍next。对比addEventListener最大的区别是事件从离散的、被动响应变成了可以被连续变换的流后面所有防抖、过滤、组合都是在这个流上继续做文章。2.2 响应式架构的三层结构源、算子、订阅端rea 在设计上将整个前端逻辑容器划分为三块这也是建议你在自己项目里复用的分法源层Source解决事件从哪里来。包括用户输入事件、DOM 事件、WebSocket 推送、定时器、ajax 请求封装、路由变化事件等统一包装成 Observable。算子层Operator解决数据如何变换。这一层只做纯函数式的转换不产生副作用比如防抖debounceTime、映射map、过滤filter、切换请求switchMap、合并流merge、状态扫描scan。业务逻辑里的大量分支判断在这里被拆解成一个个单一职责的算子链。订阅端Subscriber解决数据最终去哪。订阅端的回调里原则上只做 UI 更新、写状态库、持久化这些副作用操作。这样就保证了逻辑的可测试性算子层不碰 DOM可以单测覆盖。我画过一张图帮新人理解但这里没法画图我用文字描述一下典型链路搜索框输入事件 → 经过debounceTime(300)防抖 →map取出输入值 →filter过滤掉空白字符 →switchMap发起搜索请求 →map提取出联想结果数组 → 订阅端渲染下拉列表。每一个环节都是独立函数既是逻辑也是文档。2.3 为什么说订阅关系是响应式架构的生命线很多第一次接触响应式的人最容易忽略订阅管理。一个 Observable 被订阅后如果忘记unsubscribe回调就会一直挂着即使组件已经销毁。这跟addEventListener不 remove 是一样的问题但因为 Observable 在后台可能连着 WebSocket、定时器泄漏后果更严重。这里有一个我强烈建议的实践所有订阅都走统一的订阅管理器Subscription Bag模式。简单说就是在组件加载时创建一个Subscription容器所有流订阅都add进这个容器组件销毁时统一unsubscribe。这样从根本上解决漏掉取消订阅导致事件回调在已卸载页面上继续跑的问题。class SearchBox { constructor() { this.subs new Subscription(); // Subscription Bag } init() { const input$ fromEvent(this.inputEl, input); this.subs.add( input$ .pipe( debounceTime(300), map((e) e.target.value), switchMap((val) this.fetchSuggestions(val)) ) .subscribe((suggestions) this.renderSuggestions(suggestions)) ); } destroy() { this.subs.unsubscribe(); // 一键清理 } }只从这一点来看rea 就已经比原先散弹式清理定时器先进太多了。3. 自研还是选型站在ReactiveX的肩膀上做业务封装3.1 市面上主流响应式方案的横向对比在决定 rea 的技术栈时我们认真对比了几条路完整自研一套 Observable 实现、直接用 ReactiveX 家族的库比如 RxJS、或者退而求其次用 MobX 这类状态管理库做响应式包装。各自的取舍如下表方案编写成本算子丰富度社区资料调试工具与框架耦合度整体推荐度完全自研 Observable高核心链式调用和背压处理都要自己踩坑极低只有最基础的 map/filter无无无极不推荐RxJS 全家桶低极丰富约 100 算子庞大有插件支持低可在任何框架用强烈推荐MobX 响应式状态低无算子只做状态追踪丰富有与 React/Vue 绑定较紧场景有限传统手写回调低但后续维护高无逻辑全散落无无无不推荐作为架构当时我们也有同事提出自己写一个轻量的 observable 实现理由是用到的算子不多没必要引入几百 KB 的库。这个想法我理解但实际操作之后会发现一个致命问题背压处理。你自己实现的 Observable 只要有数据产生速度大于消费速度的场景就必须处理数据丢弃、缓存、以及通知上游减慢生产。RxJS 里debounceTime、throttleTime、bufferTime、auditTime这些算子已经处理了绝大多数情况自研版本做到debounceTime都很容易出边界 bug更别说switchMap这种带内部订阅流转的算子。3.2 rea 的做法底层用 RxJS上层做领域函数封装最终我们定的方案是基础算子直接用 RxJS但业务侧不允许直接散写复杂算子链而是要封装成语义明确的领域函数。比如import { debounceTime, map, filter, switchMap, of } from rxjs; // 把搜索联想封装成一个领域函数调用方只需要传输入流 export function createSuggestionStream(input$) { return input$.pipe( debounceTime(300), map((e) e.target.value.trim()), filter((val) val.length 0), switchMap((val) { // 这里可以在开发环境模拟延迟方便调试 return fakeFetchSuggestions(val); }) ); }这样做的收益有两个。第一业务代码和 RxJS 的解耦如果将来想换成其他响应式底层库只需要改封装层所有页面不用动。第二可读性页面负责人看createSuggestionStream的名字就知道这个流的意图而不需要逐算子推导。我在给团队定规范时明确了一条允许在任何地方使用 RxJS 算子但算子链必须被放进一个命名清晰的函数里禁止在组件内联写超过 5 个算子的链式调用。3.3 为什么没有选 MobX 而是坚持用 Observable 流如果你只在单一组件里做响应式状态MobX 确实很顺手——定义可观察对象、在组件里observer包裹、自动追踪依赖。但 rea 的场景里有大量基于事件的动态组合比如搜索防抖 请求竞态 推送合并。MobX 的处理方式是让你在 action/method 里手动管理这些时序逻辑本质还是命令式的。而 Observable 流最大的优势在于把时间维度纳入了表达范围switchMap天然表示取最后一次mergeMap表示并发保留全部这在 MobX 里需要你写额外逻辑来判断。所以 rea 最终选择了以 Observable 流为骨架把 MobX 降级为页面上少数纯展示状态的管理工具让两者各司其职。4. 三个高频场景的完整实现防抖搜索、拖拽交互与滚动加载4.1 场景一搜索联想输入用 switchMap 终结竞态问题搜索联想是 rea 应用里最常见、也最能体现响应式优势的场景。不用 rea 之前我们处理输入防抖 请求竞态要定义定时器变量和请求序号变量用 rea 之后这两个问题被debounceTime和switchMap直接吸收掉。看完整实现我简化了渲染部分重点是数据流import { fromEvent, of } from rxjs; import { debounceTime, map, filter, switchMap, catchError } from rxjs/operators; export function initSuggestionSearch() { const inputEl document.getElementById(searchInput); const listEl document.getElementById(suggestionList); const input$ fromEvent(inputEl, input); input$ .pipe( // 1. 防抖停止输入 300ms 后才触发 debounceTime(300), // 2. 取出输入框的值并去掉首尾空格 map((event) event.target.value), // 3. 过滤掉空字符串避免无意义请求 filter((value) value.trim().length 0), // 4. 这里是最关键的切换到新的内部 Observable // 当新输入来临时自动取消上一个尚未完成的请求 switchMap((value) { return fakeFetchSuggestions(value).pipe( // 5. 拉平错误避免错误把整个事件流中断 catchError((err) { console.error(搜索联想接口异常:, err); return of([]); }) ); }) ) .subscribe((suggestions) { renderSuggestionList(listEl, suggestions); }); }注意第 4 步的switchMap。它的语义是每当上游产生一个新值就订阅一个新的内部 Observable同时取消上一个内部 Observable 的订阅。等于说用户的输入以 300ms 的节奏不断产生switchMap会切断前一次还没回来的请求只保留最新这次的结果。这个行为正好就是我们所需的搜索竞态防护而且不需要我们手写任何序号变量。我实测下来的体验是快速输入rea这个词的时候真实发出的网络请求只有r、re、rea这三个中间的最后一个具体取决于打字停顿节奏但绝不会出现r 的结果晚于 rea 到达从而覆盖列表的情况。这对用户体感的提升非常明显。4.2 场景二拖拽排序把按下/移动/松开拆成三段流再合并拖拽交互如果用命令式写法需要分别注册mousedown、mousemove、mouseup三个事件然后在全局状态里保存是否正在拖拽的布尔值移动时判断如果正在拖拽才更新位置。这类写法的问题是全局标志位容易被其他逻辑干扰比如某个弹层打开时鼠标抬起事件没收到就会一直拖拽。rea 的处理方式不同把鼠标事件拆成独立的流再用groupBy或者闭包方式把它们按一次拖拽会话组合起来。import { fromEvent, merge } from rxjs; import { takeUntil, switchMap, map, tap } from rxjs/operators; export function initDragSort(listEl) { const mouseDown$ fromEvent(listEl, mousedown); const mouseMove$ fromEvent(document, mousemove); const mouseUp$ fromEvent(document, mouseup); // 对于每个 mousedown创建一个独立的拖拽会话流 mouseDown$ .pipe( switchMap((downEvent) { const targetRow downEvent.target.closest(.row); if (!targetRow) return []; const startX downEvent.clientX; const startY downEvent.clientY; const originTop targetRow.offsetTop; // 在当前会话内持续监听 mouseMove直到 mouseUp 为止 return mouseMove$.pipe( map((moveEvent) ({ targetRow, offsetY: moveEvent.clientY - startY, originTop, })), // 关键mouseUp 发生时结束当前会话流 takeUntil(mouseUp$) ); }) ) .subscribe(({ targetRow, offsetY, originTop }) { // 订阅端只负责更新 UI不参与事件流组合逻辑 targetRow.style.transform translateY(${originTop offsetY}px); }); }这段代码的核心价值在于takeUntil(mouseUp$)它的作用是当 mouseUp 事件发生时自动终止当前这个拖拽会话序列。这样你不需要手动清理 mousemove 的监听也不会出现松手后还在拖拽的残留状态。从事件流的角度看一次拖拽被完整建模为一次 mousedown 到下一次 mouseup 之间的所有 mousemove语义非常清晰。4.3 场景三滚动触底加载用 scan 维护分页状态无限滚动加载里最容易写乱的是页码状态。命令式写法通常是滚动到底→pageNum1→发请求→拼接数据中间要处理加载中锁、失败回退页码等边界。rea 的写法是用scan算子把每一次滚动事件折叠成一条页码变化史。import { fromEvent } from rxjs; import { filter, throttleTime, scan, mergeMap, tap, catchError, of } from rxjs/operators; export function initInfiniteScroll() { const scroll$ fromEvent(window, scroll); scroll$ .pipe( // 滚动事件过于密集用 throttleTime 限制触发频率 throttleTime(200), // 判断是否触底页面滚动到底部 300px 内 filter(() window.innerHeight window.scrollY document.body.offsetHeight - 300), // scan 像是一个会记住上一次结果的 reduce这里记录当前页面序号 scan((page, _event) page 1, 0), // 用 mergeMap 允许并发请求或者用 concatMap 保证顺序加载 mergeMap((page) fetchNextPage(page).pipe( tap((res) { appendData(res.list); }), catchError((err) { console.error(加载失败保留页码以便手动重试, err); return of(null); }) ) ) ) .subscribe(); }scan算子在这个场景里扮演的角色是内部状态折叠器每触发一次滚动触底它就基于上一次的页码加一并且把这个值传给下游。这样就不存在在外部维护 pageNum 变量的烦恼多个订阅之间天然隔离因为状态是闭包在算子里面的。值得注意的是如果用户滚动特别快mergeMap会产生叠加请求如果希望严格按顺序加载换成concatMap即可。这个替换只需要改一个算子名命令式写法想改并发策略要动一大片逻辑这就是 rea 在演进灵活性上的显著收益。5. 项目落地后的四个深坑竞态、错误中断、背压与调试难点5.1 坑一初学最易误用的 mergeMap 与 switchMap 选型rea 项目第一个大坑出现在团队同学关于搜索和加载的并发语义理解上。不少同学在写搜索联想时用了mergeMap而不是switchMap。mergeMap的语义是上游每一个值创建的内部 Observable都会保留并发执行最后所有结果合并输出。如果用户快速输入多个搜索词mergeMap会导致多次请求同时进行且返回顺序不确定先发出的请求可能后返回、进而覆盖后发请求的结果——这就是经典的竞态 bug。switchMap的语义是上游新值到来时取消上一个内部 Observable 的执行切换到新值对应的内部 Observable。它天然屏蔽了旧结果。简而言之算子并发策略适用场景switchMap最新优先取消旧订阅搜索联想、路由切换mergeMap保留全部并发多个无关联异步任务同时发起concatMap串行按顺序排队表单提交按序保存、逐张上传exhaustMap当前未完成则忽略新值登录按钮防连点我总结的团队口诀是最新消息要覆盖旧状态用 switchMap事件之间有先后强依赖用 concatMap独立事件并发做用 mergeMap用户疯狂操作但要保护资源用 exhaustMap。每次 code review 看到不合理的 map 算子我都会要求对方说出选择理由说不出来就默认用 switchMap。5.2 坑二错误处理丢一赔全一个请求失败导致整条事件流断掉第二个坑非常隐蔽但后果严重。Observable 流有一个特性只要内部抛错并传给error回调就代表这条流直接终结后面不会再有任何数据。这带来一个后果——如果你在一串算子里忘记加catchError某个接口超时抛错那么整条搜索输入流就永久死亡用户再输入任何字都不会触发搜索了而且控制台大概率没有明显报错很难察觉。我记录过一次真实的故障排查过程。现象是某个列表页的筛选条件在用户切换几次后就完全没反应了。我们最开始怀疑是筛选组件的数据绑定问题后来逐步加日志排查发现不是 UI 层的问题而是筛选联动的事件流在第一次请求因网络超时抛错后整个流就中断了。具体链路用户切换筛选选项 →change事件进入流。mergeMap发起请求第一次请求因某个参数导致后端 500。错误没有被捕获直接传给订阅端errorObservable 终止。订阅端没有写error回调控制台甚至看不到错误。后续的筛选事件持续产生但流已经终结不会触发任何下游逻辑。修复方式是在容易出错的请求处统一加catchError并且返回一个可继续操作的空值流保证流不断掉import { of, throwError } from rxjs; import { catchError } from rxjs/operators; // 错误不中断流而是转为一个错误通知 空结果 .filterChange$ .pipe( debounceTime(100), switchMap((filterValue) fetchFilterResult(filterValue).pipe( catchError((err) { console.error(筛选请求失败已忽略此错误以保流存活, err); return of({ list: [], total: 0 }); }) ) ) ) .subscribe((result) renderResult(result));这个经验后来被写进了团队规范任何 Observable 链路的末端都必须有 catchError 或者 error 回调否则视为缺陷。5.3 坑三上游高频事件与背压——快速滚动时的意外内存飙升背压这个概念听起来很高大上其实落到浏览器场景里就是数据产生的速度比消费的速度快很多时中间积压的事件流把内存撑爆了。我们在滚动加载场景里踩过一个很典型的坑滚动事件在浏览器里每帧可以触发多次如果不加节流生成的事件流以每秒钟几十到上百的速度往下游灌。下游的触底判断和请求逻辑本身不慢但在快速滚动时scan折叠的页码会瞬间飙到几十mergeMap同时发出去十几个请求等待返回期间页面卡顿明显甚至掉帧。解决方案是在源头加throttleTime或auditTime而不是在需求出现后才补救。我开始实现时用throttleTime(200)限制了滚动事件频率又配合exhaustMap保证在上一个请求未完成时忽略后续触底事件双保险之后内存和请求数都正常了。这也是 rea 架构里的一个重要原则算子链越靠上游对性能的影响越呈指数级放大。源头一个节流算子的作用远大于下游十个判断条件。5.4 坑四调试困难——事件什么时候发、为什么没发很难追响应式代码的语义虽然清晰但调试时比命令式代码更难你没法在各步骤之间打断点直观看到数据结构。命令式代码出 bug 可以逐行单步Observable 流的每个算子都是内部封装好的单步调试进去全是库源码毫无意义。我们的解法是事件标签 日志中间件 状态快照三件套。第一每个重要流在创建时加一个tag前缀日志打点用它标识来源第二封装一个log自定义算子在关键链路节点输出流的值和时间戳第三订阅端统一通过tap打印最终数据。核心代码如下import { Observable } from rxjs; import { tap } from rxjs/operators; export function debugLog(tag, level info) { return tap((value) { console[level]([${new Date().toISOString()}] [${tag}], value); }); } // 使用示例 input$.pipe( debugLog(search-input), debounceTime(300), debugLog(search-after-debounce), switchMap((value) fetchSuggestions(value)), debugLog(search-response) );这样在排查问题时只要看控制台的时间线就能确认每个阶段是否触发、数据长什么样、哪个算子的输出不符合预期。比逐行打断点高效得多。我还建议把日志输出格式统一成[时间] [标签] 数据内容这样能在日志聚合系统里按标签搜索快速定位问题流。6. 关于扩展性与 debounce 之外的一些实战补充6.1 不只处理事件WebSocket 推送消息也可以建模成 Observablerea 的架构不应局限于 DOM 事件。我们后期把 WebSocket 推送也统一包装成了 Observable这样推送消息和其他事件流可以通过merge、combineLatest做组合联动。一个实操例子是页面同时监听新工单推送和当前筛选条件当推送的新工单满足当前筛选条件时用combineLatest自动触发列表更新而不需要在推送回调里去判断当前筛选选择了什么。import { combineLatest } from rxjs; import { map } from rxjs/operators; // socket 消息流{ type: new-ticket, data: {...} } // 筛选条件流{ priority: high, status: open } const newTicket$ ticketSocketStream$().pipe( filter((msg) msg.type new-ticket), map((msg) msg.data) ); const filter$ currentFilter$; // 当两者任意变化时自动计算是否需要插入列表 combineLatest([newTicket$, filter$]) .pipe( map(([ticket, filter]) ({ ticket, shouldInsert: matchFilter(ticket, filter), })) ) .subscribe(({ ticket, shouldInsert }) { if (shouldInsert) { insertTicketAtTop(ticket); } });这个模式在聊天类应用、实时看板、监控大屏都是标配。把推送流建模成 Observable 之后新增一种消息类型只需要在源层加一个filter分支不需要在组件里增加新的消息监听和清理逻辑。6.2 注意 unsubscribe 的完整性和时机前面提到订阅要统一管理但还有一个细节unsubscribe 并不能解决所有异步副作用。如果 Observable 内部发起了一个 fetch 请求比如from(fetch(...))取消订阅只是说下游不再接收结果但网络请求本身已经发出去了无法取消除非用 AbortController。这意味着如果你用switchMap在每次输入时发起搜索请求旧请求仍会在后台实际发出并完成只是结果被丢弃了。这算是一个需要接受的成本竞态防护解决的是结果是否正确显示的问题而不是请求是否发出的问题。在高频场景下如果后端接口性能敏感建议在封装层加 AbortController 的取消逻辑或者让后端配合做去重。6.3 rea 后续演进的两个方向项目进入稳定期后我们主要往两个方向演进。第一个是可视化流图调试面板用自定义操作符收集每个流的标签和值绘制出开发时的事件流图让业务同学也能直观看到数据流向。第二个是流式分层管理从单一页面级 rea 延伸出跨页面的共享流层把用户登录态、全局配置、通知中心都作为全局流注入每个页面减少重复订阅。用户体验上也有一些小细节值得分享。比如加载状态也可以通过一个独立的 Observable 流来推导而不是用布尔变量将请求开始和请求结束映射成true和false再合并所有请求流来驱动页面的加载动画能避免多个请求并发时状态互相覆盖的问题。这个思路我建议所有用 rea 的朋友都试一下效果非常直接。7. 一点感悟响应式架构是约束不是银弹最后说回这个项目本身。很多团队引入响应式架构最终挫败感很强问题通常出在两个地方一是没有定清楚规范让所有人自由发挥最后算子链满天飞、流多到自己也看不懂二是没有把流当成显式的架构层来管理而是把 Observable 当成了一个加强版的 Promise 工具。rea 之所以能成功并不在于响应式的概念有多新、多高级而在于它逼着我们做了几个很朴素的约束任何异步源必须包装成流任何算子链必须封装成语义明确的函数任何订阅必须有对应的销毁时机任何进入订阅端的错误必须显式处理。这四个约束在执行层面其实很简单但它们解决了我们从项目第一天就头疼的时序乱、状态散、清理难的问题。如果你现在正被一个满是回调、竞态、散乱状态标识的页面折磨我建议别急着上微前端、别急着搞复杂的状态库先拿一个相对独立的模块用 Observable 流把事件源梳理一遍。你会发现写着写着很多以前靠自觉才能保持的代码秩序会变成架构自带的东西。这就是 rea 给我最大的收获。