RxJS Observer 完全指南理解 next、error、complete 三类通知回调与订阅机制【免费下载链接】rxjsA reactive programming library for JavaScript项目地址: https://gitcode.com/gh_mirrors/rx/rxjsObserver观察者是 RxJS 响应式编程模型中消费数据的一端它是一组回调函数的集合分别对应 Observable 在每次执行中可能投递的三类通知——next、error与complete。本篇指南以 apps/rxjs.dev/content/guide/observer.md 为核心骨架结合当前仓库packages/rxjs的源码实现完整讲解 Observer 的定义、完整/部分两种形态、三种订阅传参方式以及 Observable 契约对通知序列的约束。读完后你将能够准确写出符合 RxJS 约定的 Observer并理解observable.subscribe(...)内部是如何把各种参数形态归一化为 Observer 的。Observer 是什么Observer 是 Observable 所投递值的消费者。它本质上只是一组回调的集合每种回调对应 Observable 投递的一种通知类型next接收一个值Number、String、Object 等任意 JavaScript 值error接收一个 JavaScript Error 或异常complete不携带任何值仅表示数据流结束。一个典型的 Observer 对象如下const observer { next: x console.log(Observer got a next value: x), error: err console.error(Observer got an error: err), complete: () console.log(Observer got a complete notification), };使用它的方式很简单——把它作为参数传给某个 Observable 的subscribeobservable.subscribe(observer);Observer 就是包含三个回调的普通对象每个回调对应 Observable 可能投递的一种通知。部分 ObserverPartial ObserverRxJS 中的 Observer 可以是不完整的partial。如果你省略了某个回调Observable 的执行依然会正常进行只是没有对应回调的那些通知会被忽略。例如下面这个 Observer 没有提供complete回调const observer { next: x console.log(Observer got a next value: x), error: err console.error(Observer got an error: err), };这种缺省即忽略的设计在源码层面同样体现得很直接packages/rxjs/src/tap.ts中定义的操作符 ObserverTapObserverT extends ObserverT其内部正是通过可选调用tapObserver.next?.(value)、tapObserver.error?.(sourceError)、tapObserver.complete?.()来兼容只提供部分回调的用法packages/rxjs/src/notification.ts中Notification.accept的入参类型也写为PartialObserverT | ((value: T) void)说明整个库对部分 Observer是普遍接纳的。直接传入 next 回调订阅时你还可以只传一个next回调作为参数而不必显式构造 Observer 对象observable.subscribe(x console.log(Observer got a next value: x));在observable.subscribe内部RxJS 会用这个回调参数创建出一个 Observer 对象并把该回调作为其next处理器。也就是说上面这条语句在语义上等价于observable.subscribe({ next: x console.log(Observer got a next value: x) });源码印证订阅参数如何归一化当前仓库packages/rxjs/src/cold-observable.ts中的ColdObservable.subscribe给出了这一归一化过程的实现证据subscribe(maybeObserver?: PartialObserverT | ((value: T) void) | null, config?: SubscribeOptions): void { const subscriber new ColdSubscriber(maybeObserver, config?.signal); try { this.#init(subscriber); } catch (error) { subscriber.error(error); } }而ColdSubscriber的构造逻辑同文件第 2026 行附近正是文档所述用回调参数创建 Observer 对象的落地代码constructor(maybeObserver: PartialObserverT | ((value: T) void) | null null, parentSignal?: AbortSignal) { // 当传入的是函数时包装为 { next: maybeObserver } 形式的 Observer typeof maybeObserver function ? { next: maybeObserver } : maybeObserver; // ... }由此可见subscribe接受三种参数形态完整 Observer 对象、部分 Observer 对象、或单个 next 函数最终都会被统一处理。另外注意到该签名中的SubscribeOptions如{ signal }还支持传入AbortSignal以支持中止订阅这与subscription.unsubscribe()一样是取消执行的一种途径。Observer 与 Subject 的关系Observer 不仅仅用于subscribe它也构成 RxJS 双向类型体系的一部分。在 packages/rxjs/src/util/types.ts 中可以看到export interface SubjectLikeIn, Out In extends SubscribableOut, ObserverIn { readonly active: boolean; }SubjectLike以及仓库中的AsyncSubject、BehaviorSubject等 Subject 家族同时实现了可订阅源与Observer两个身份——这正是 Subject 既能被订阅、又能通过next/error/complete向外界推送数据的类型基础。理解这一点有助于在阅读 Subject 相关源码时把 Observer 的三个回调当作 Subject 的输入接口来理解。通知类型与 Observable 契约虽然本篇聚焦 Observer但要真正用好三个回调必须理解 Observable 执行Observable Execution对通知序列的约束即所谓的Observable 契约 / 文法用正则表达式表达为next*(error|complete)?含义是可以投递零到无限次next通知一旦投递了error或complete其后不得再投递任何通知error与complete二者只能出现其中一个且至多一次。违反契约的例子下面这段代码中complete()之后的next(4)不会被投递因为它违反了契约import { Observable } from rxjs; const observable new Observable(function subscribe(subscriber) { subscriber.next(1); subscriber.next(2); subscriber.next(3); subscriber.complete(); subscriber.next(4); // 不会被投递违反契约 });用 try/catch 保证 error 通知Observable 严格遵循该契约一个良好的实践是在subscribe函数体内用try/catch包裹把捕获到的异常转成error通知投递给 Observerimport { Observable } from rxjs; const observable new Observable(function subscribe(subscriber) { try { subscriber.next(1); subscriber.next(2); subscriber.next(3); subscriber.complete(); } catch (err) { subscriber.error(err); // 捕获到异常时投递 error 通知 } });这种写法在源码中也有对应实现packages/rxjs/src/cold-observable.ts的subscribe方法把初始化函数this.#init(subscriber)包在try中一旦抛出异常就调用subscriber.error(error)从库的层面保证 Observer 总能收到正确的error通知。继续深入与 Observable、Subscription 的关系Observer 是理解 RxJS 数据流的关键一环它与另外两个核心概念紧密相连Observable数据的生产者负责投递通知Observer 则是通知的接收方。二者共同构成 Push 模型的两端。参见 apps/rxjs.dev/content/guide/observable.md。Subscriptionobservable.subscribe(...)返回的执行句柄调用subscription.unsubscribe()可以取消正在进行的执行并释放资源。参见 apps/rxjs.dev/content/guide/subscription.md。小结Observer 是包含next、error、complete三个回调的普通对象分别消费值、异常与结束通知三种回调都可以省略省略的通知类型会被忽略Observable 执行不受影响订阅时可以传完整 Observer、部分 Observer或直接传单个next函数——内部会自动包装为 ObserverObservable 严格执行next*(error|complete)?契约error/complete之后不会再有任何通知建议在 Observable 的subscribe函数体内用try/catch把异常转为error通知这一做法与库自身实现如ColdObservable.subscribe中的 try/catch保持一致。把握住Observer 三个回调这一朴素本质再看 RxJS 的操作符、Subject 与高阶数据流就会清晰很多。【免费下载链接】rxjsA reactive programming library for JavaScript项目地址: https://gitcode.com/gh_mirrors/rx/rxjs创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 RxJS Observer 完全指南：理解 next、error、complete 三类通知回调与订阅机制
RxJS Observer 完全指南理解 next、error、complete 三类通知回调与订阅机制【免费下载链接】rxjsA reactive programming library for JavaScript项目地址: https://gitcode.com/gh_mirrors/rx/rxjsObserver观察者是 RxJS 响应式编程模型中消费数据的一端它是一组回调函数的集合分别对应 Observable 在每次执行中可能投递的三类通知——next、error与complete。本篇指南以 apps/rxjs.dev/content/guide/observer.md 为核心骨架结合当前仓库packages/rxjs的源码实现完整讲解 Observer 的定义、完整/部分两种形态、三种订阅传参方式以及 Observable 契约对通知序列的约束。读完后你将能够准确写出符合 RxJS 约定的 Observer并理解observable.subscribe(...)内部是如何把各种参数形态归一化为 Observer 的。Observer 是什么Observer 是 Observable 所投递值的消费者。它本质上只是一组回调的集合每种回调对应 Observable 投递的一种通知类型next接收一个值Number、String、Object 等任意 JavaScript 值error接收一个 JavaScript Error 或异常complete不携带任何值仅表示数据流结束。一个典型的 Observer 对象如下const observer { next: x console.log(Observer got a next value: x), error: err console.error(Observer got an error: err), complete: () console.log(Observer got a complete notification), };使用它的方式很简单——把它作为参数传给某个 Observable 的subscribeobservable.subscribe(observer);Observer 就是包含三个回调的普通对象每个回调对应 Observable 可能投递的一种通知。部分 ObserverPartial ObserverRxJS 中的 Observer 可以是不完整的partial。如果你省略了某个回调Observable 的执行依然会正常进行只是没有对应回调的那些通知会被忽略。例如下面这个 Observer 没有提供complete回调const observer { next: x console.log(Observer got a next value: x), error: err console.error(Observer got an error: err), };这种缺省即忽略的设计在源码层面同样体现得很直接packages/rxjs/src/tap.ts中定义的操作符 ObserverTapObserverT extends ObserverT其内部正是通过可选调用tapObserver.next?.(value)、tapObserver.error?.(sourceError)、tapObserver.complete?.()来兼容只提供部分回调的用法packages/rxjs/src/notification.ts中Notification.accept的入参类型也写为PartialObserverT | ((value: T) void)说明整个库对部分 Observer是普遍接纳的。直接传入 next 回调订阅时你还可以只传一个next回调作为参数而不必显式构造 Observer 对象observable.subscribe(x console.log(Observer got a next value: x));在observable.subscribe内部RxJS 会用这个回调参数创建出一个 Observer 对象并把该回调作为其next处理器。也就是说上面这条语句在语义上等价于observable.subscribe({ next: x console.log(Observer got a next value: x) });源码印证订阅参数如何归一化当前仓库packages/rxjs/src/cold-observable.ts中的ColdObservable.subscribe给出了这一归一化过程的实现证据subscribe(maybeObserver?: PartialObserverT | ((value: T) void) | null, config?: SubscribeOptions): void { const subscriber new ColdSubscriber(maybeObserver, config?.signal); try { this.#init(subscriber); } catch (error) { subscriber.error(error); } }而ColdSubscriber的构造逻辑同文件第 2026 行附近正是文档所述用回调参数创建 Observer 对象的落地代码constructor(maybeObserver: PartialObserverT | ((value: T) void) | null null, parentSignal?: AbortSignal) { // 当传入的是函数时包装为 { next: maybeObserver } 形式的 Observer typeof maybeObserver function ? { next: maybeObserver } : maybeObserver; // ... }由此可见subscribe接受三种参数形态完整 Observer 对象、部分 Observer 对象、或单个 next 函数最终都会被统一处理。另外注意到该签名中的SubscribeOptions如{ signal }还支持传入AbortSignal以支持中止订阅这与subscription.unsubscribe()一样是取消执行的一种途径。Observer 与 Subject 的关系Observer 不仅仅用于subscribe它也构成 RxJS 双向类型体系的一部分。在 packages/rxjs/src/util/types.ts 中可以看到export interface SubjectLikeIn, Out In extends SubscribableOut, ObserverIn { readonly active: boolean; }SubjectLike以及仓库中的AsyncSubject、BehaviorSubject等 Subject 家族同时实现了可订阅源与Observer两个身份——这正是 Subject 既能被订阅、又能通过next/error/complete向外界推送数据的类型基础。理解这一点有助于在阅读 Subject 相关源码时把 Observer 的三个回调当作 Subject 的输入接口来理解。通知类型与 Observable 契约虽然本篇聚焦 Observer但要真正用好三个回调必须理解 Observable 执行Observable Execution对通知序列的约束即所谓的Observable 契约 / 文法用正则表达式表达为next*(error|complete)?含义是可以投递零到无限次next通知一旦投递了error或complete其后不得再投递任何通知error与complete二者只能出现其中一个且至多一次。违反契约的例子下面这段代码中complete()之后的next(4)不会被投递因为它违反了契约import { Observable } from rxjs; const observable new Observable(function subscribe(subscriber) { subscriber.next(1); subscriber.next(2); subscriber.next(3); subscriber.complete(); subscriber.next(4); // 不会被投递违反契约 });用 try/catch 保证 error 通知Observable 严格遵循该契约一个良好的实践是在subscribe函数体内用try/catch包裹把捕获到的异常转成error通知投递给 Observerimport { Observable } from rxjs; const observable new Observable(function subscribe(subscriber) { try { subscriber.next(1); subscriber.next(2); subscriber.next(3); subscriber.complete(); } catch (err) { subscriber.error(err); // 捕获到异常时投递 error 通知 } });这种写法在源码中也有对应实现packages/rxjs/src/cold-observable.ts的subscribe方法把初始化函数this.#init(subscriber)包在try中一旦抛出异常就调用subscriber.error(error)从库的层面保证 Observer 总能收到正确的error通知。继续深入与 Observable、Subscription 的关系Observer 是理解 RxJS 数据流的关键一环它与另外两个核心概念紧密相连Observable数据的生产者负责投递通知Observer 则是通知的接收方。二者共同构成 Push 模型的两端。参见 apps/rxjs.dev/content/guide/observable.md。Subscriptionobservable.subscribe(...)返回的执行句柄调用subscription.unsubscribe()可以取消正在进行的执行并释放资源。参见 apps/rxjs.dev/content/guide/subscription.md。小结Observer 是包含next、error、complete三个回调的普通对象分别消费值、异常与结束通知三种回调都可以省略省略的通知类型会被忽略Observable 执行不受影响订阅时可以传完整 Observer、部分 Observer或直接传单个next函数——内部会自动包装为 ObserverObservable 严格执行next*(error|complete)?契约error/complete之后不会再有任何通知建议在 Observable 的subscribe函数体内用try/catch把异常转为error通知这一做法与库自身实现如ColdObservable.subscribe中的 try/catch保持一致。把握住Observer 三个回调这一朴素本质再看 RxJS 的操作符、Subject 与高阶数据流就会清晰很多。【免费下载链接】rxjsA reactive programming library for JavaScript项目地址: https://gitcode.com/gh_mirrors/rx/rxjs创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考