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

Question

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

我考虑过为此使用主题 - 这就是它们的用途：

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 下一个功能正常。

这向我表明，在观察者、可观察对象和主题等方面，有些 RxJS 是我还没有理解的。一般来说，我了解如何将事件源和其他类似的东西连接到可观察的流中。它实际上只是在从一个无聊的函数调用中构建一个可观察流的上下文中 - 每当外部库调用该函数时，这个流都会发出一个更新的 observable 信号。

Answer 1

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

所以，如果我给你一个简单的观察者，你可以打电话给next()/error()/complete() 以任何顺序，随意多次，即使它会是 在你打电话给我的complete() 之后，你打电话给我的next() 太糟糕了。

另一方面，如果我给你一个观察者包裹在一个 订阅者，现在有状态，如果您尝试拨打next() on 在你打电话给complete() 之后的那个订阅者，我不会看到它。 如果你打电话给unsubscribe()，我就超脱了。

当你调用 subscribe 时，如

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() 接口，让您可以为运算符链的前端提供数据，
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?。