HOWTO · JavaScript
JavaScript subscribe()方法
如何使用不同示例的 JavaScript 代码语句中的 subscribe() 方法创建和执行订阅。
本页内容
本文将介绍 RxJS 如何通过 subscribe() 启动 Observable 执行、如何处理通知，以及如何通过 unsubscribe() 停止工作并释放资源。jQuery 事件插件也可能使用 subscribe() 这个名称，但那是行为不同的另一套 API。
订阅 RxJS Observable
Observable 描述了如何生成值。调用 subscribe() 会为该订阅者启动一次执行，并返回一个 Subscription 对象。这个对象代表该次执行，并提供停止执行的方式。再次调用 subscribe() 会创建一个独立的订阅；它并非只是给共享的 Promise 再附加一个回调。
在 RxJS 中，常见的数据源是 interval(1000)，它每秒发出一个递增的数字。例如，const subscription = interval(1000).subscribe(value => console.log(value)); 会启动这项工作。下面的无依赖教学模型展示了通知和 teardown 生命周期；它不是 RxJS 的实现。正常完成、发生错误，或对活动中的订阅显式取消订阅后，teardown 都只会运行一次。
运行此示例：
本节中的代码以及显示的标签或输出仍为英文。
function observable(produce) {
return {
subscribe(observer) {
let closed = false;
let teardown;
let teardownRequested = false;
let teardownRan = false;
function runTeardown() {
if (teardownRequested && teardown && !teardownRan) {
teardownRan = true;
teardown();
}
}
function close() {
if (!closed) {
closed = true;
teardownRequested = true;
runTeardown();
}
}
const safeObserver = {
next(value) {
if (!closed) observer.next(value);
},
error(error) {
if (!closed) {
close();
if (observer.error) observer.error(error);
}
},
complete() {
if (!closed) {
close();
if (observer.complete) observer.complete();
}
}
};
teardown = produce(safeObserver);
runTeardown();
return { unsubscribe: close };
}
};
}
const source = observable(observer => {
observer.next("first");
observer.next("second");
observer.complete();
return () => console.log("cleanup");
});
source.subscribe({
next: value => console.log("value:", value),
error: error => console.log("error:", error.message),
complete: () => console.log("complete")
});
let errorCleanupCount = 0;
observable(observer => {
observer.error(new Error("source failed"));
return () => { errorCleanupCount += 1; };
}).subscribe({ error: () => {} });
if (errorCleanupCount !== 1) throw new Error("error teardown did not run once");
let cancelCleanupCount = 0;
const active = observable(() => () => { cancelCleanupCount += 1; }).subscribe({});
active.unsubscribe();
active.unsubscribe();
if (cancelCleanupCount !== 1) throw new Error("unsubscribe teardown did not run once");
value: first
value: second
complete
cleanup
输出展示了正常完成的路径。额外检查覆盖了错误和显式取消订阅两种情况，表明每条路径都会调用一次 teardown；重复取消订阅也不会造成问题。在 RxJS 中，库提供的 Subscription 承担资源管理的职责；在应用代码中，请使用 RxJS 的 Observable 和 Subscription。
处理 Observable 通知
观察者可以提供三个回调：每次收到值时调用 next，发生故障时调用 error，数据源正常结束时调用 complete。它们是不同的通知，并非 Promise 方法。发生错误会结束该次执行；只有数据源完成时才会发送完成通知，因此这两个回调都不表示“无论如何都会运行”。如果数据源始终不完成，它可以持续发出值，直到被取消订阅。
在 RxJS 中，可以将观察者对象传给 subscribe()，例如 { next: value => handle(value), error: error => report(error), complete: () => finish() }。如果应用需要处理数据源故障，请提供错误处理程序。让回调中的工作保持精简，并记住值可能会多次到达。
取消订阅并分组管理订阅
不再需要这项工作时，调用 subscription.unsubscribe()。取消订阅会停止该订阅的通知，并运行数据源注册的 teardown。它不会撤销已经完成的工作，也不一定会停止仍有其他活动订阅者使用的共享数据源。
例如，先启动 const timer = interval(1000).subscribe(value => console.log(value));，计时器不再需要时，再调用 timer.unsubscribe()。Subscription 还可以管理子订阅：用 new Subscription() 创建父订阅，通过 parent.add(child) 添加每个子订阅，再调用 parent.unsubscribe() 清理整组订阅。父子关系也会影响添加时机：如果父订阅已经取消，再添加子订阅，该子订阅也会随之取消。添加子订阅和之后取消整组订阅时，请保持使用同一个父订阅变量。
使用 jQuery 的 subscribe() 插件
某些 jQuery 事件聚合插件会提供 subscribe(topic, handler) 函数。在这类 API 中，topic 是事件名称，处理程序接收事件数据；它不是 RxJS 的 Observable 订阅。现有资料没有指出插件或软件包的名称，因此无法在此核实它的确切调用签名、配置方式和清理方法。
在确认插件文档之前，应将 subscribe("HelloWorld", handler) 这样的调用视为概念示例。使用前请查明该插件如何发布 topic，以及如何移除处理程序。不要把它的事件名称或清理规则与 RxJS 的 Subscription 对象混为一谈。