rxjs - 使用 subscription.next() 是反模式吗？

Question

我有一个订阅，我想在一段中间件代码中使用它来发出事件信号。所以我有一个东西想要订阅一个可观察的，另一个东西不是一个可观察的。

我考虑过为此使用 Subjects - 这就是它们的用途:

const repeater = new Rx.Subject();

function subscribe(observer) {
  return repeater.subscribe(observer);
}

// in some other function, call repeater.next(val)

但后来我开始查看常规 subscribe() 调用返回的内容。我可以这样做:

let sub = null;
function subscribe(observer) {
  return Rx.Observable.create((o) => sub = o);
}
// elsewhere sub.next(val)

还有？

let unsub = null;
function subscribe(observer) {
  unsub = Rx.Observable.create(() => false).subscribe(observer)
}
// elsewhere unsub.next(val)

所有这些都会向订阅者发送 val 信号。我在这里不明白的奇怪的事情是订阅返回有一个可用的 next() - 我认为 next() 只存在于 Observable 的观察者身上上下文。

无论如何，我都需要处理取消订阅的小玩意——当中间件被拆除时，我需要发出流完成信号并释放一些资源。令我惊讶的是，unsub 的下一个功能。

这向我发出信号，就 Observers 和 Observables 以及 Subjects 等而言，有一些我还没有理解的 RxJS。总的来说，我了解如何将事件源和其他类似的东西连接到可观察的流中。它实际上只是在从无聊的函数调用构建可观察流的上下文中——每当外部库调用该函数时，该流都会发出更新的可观察对象的信号。

Answer 1

订阅者扩展订阅和观察者，添加状态。它公开了一种更改状态的方法(即 unsubscribe())，以及它还公开了观察者的 next()/error()/complete() 方法，但是这些方法现在既支持状态又改变状态。

所以，如果我给你一个裸观察者，你可以调用 next()/error()/complete()以任何顺序，你想要的次数都可以，即使它会在调用我的 complete() 之后调用我的 next() 太糟糕了。

另一方面，如果我给你一个包裹在订阅者，现在有状态，如果你尝试调用 next()在你调用 complete() 之后的那个订阅者，我不会看到它。如果你调用 unsubscribe()，我将脱离。

当您调用订阅时，如

subscriber = Rx.Observable.create(fn).subscribe(observer);

你得到的是同一个观察者，而且只有那个观察者，包裹在订阅者中。这就是您看到 next()/error()/complete() 方法的原因。但是这些方法通常供内部使用，如果您使用它们来喂养观察者，它不会做您期望的事情:

let observerA = {
    next: (x) => console.log('A: value: ' + x),
    error: (x) => console.log('A: error: ' + x),
    complete: () => console.log('A: completed')
}
let observerB = {
    next: (x) => console.log('B: value: ' + x),
    error: (x) => console.log('B: error: ' + x),
    complete: () => console.log('B: completed')
}

let observable = Rx.Observable.create(() => false);
let subscriberA = observable.subscribe(observerA);
let subscriberB = observable.map(x => 10*x).subscribe(observerB);
subscriberA.next(1); // only feeds observerA
// => "A: value: 1"
subscriberB.next(2); // only feeds observerB
// => "B: value: 2"  // What?

很有可能，对于您的用例，您会

1. 想要使用 Subject，因为它为您提供了 next()/error()/complete() 接口(interface)，让您可以提供前端信息运营商链，
2. 想要使用 Subject，因为它可以让您向多个观察者提供相同的值，
3. 忘记你刚刚学到的有关订阅者的知识，因为你不会使用 next()/error()/complete() 用户界面。相反，将 subscribe() 返回的对象仅视为订阅，并且仅对其使用 unsubscribe() 方法。

所以:

let subject = new Rx.Subject();
let subscriptionA = subject.subscribe(observerA);
let subscriptionB = subject.map(x=>10*x).subscribe(observerB);
subject.next(3);
// => A: value: 3
// => B: value: 30
subscriptionA.unsubscribe()
subject.next(4);
// => B: value: 40
subscriptionB.unsubscribe()
subject.next(5);
// (no output)

另见 When to use asObservable() in rxjs? .