订阅 observables 的主题

Question

我有以下简单代码：

let a: Observable<any> = Observable.from([1,2,3,4,5]).share();
let b: Subject<any> = new Subject();

a.subscribe(b)
a.subscribe(value => console.log("New observable value: " + value));
b.subscribe(value => console.log("New subject value: " + value));

b.next("value1");
b.next("value2");

我看到的唯一控制台行是对于数组中的每个值都带有“新可观察值：”的控制台行。如果我注释掉第 3 行，那么我会得到所有的“新可观察值”行和“新主题值”行，但主题当然不会得到可观察值（数组的一部分）省略的任何值。

这么基本的问题 - 为什么会这样？为什么我的主题没有看到任何数组元素 1、2、3、4、5？它订阅 observable，所以它应该得到这些事件（share() 应该广播给所有订阅者），但这似乎是我的一个误解。

我在这里寻找的是对该主题进行一次订阅，并查看来自可观察对象的所有值以及通过调用next() 发出的任何值。是我的代码错误，还是我试图以错误的方式做事？

Answer 1

您的基本问题是订阅顺序，当您订阅 console.log 时，Subject 已经收到数据。 （查看下面的代码示例以查看它的运行情况）

第二个问题，为什么next("value1") 不起作用，是因为a.subscribe(b) 在b 发射后也会在b 上调用.complete()。

let a = Rx.Observable.from([1,2,3,4,5]).share();
let b = new Rx.Subject();

b.subscribe(value => console.log("New subject value: " + value));
a.subscribe(value => console.log("New observable value: " + value));
a.subscribe(b);

<script src="https://unpkg.com/rxjs/bundles/Rx.min.js"></script>

---

一种解决此问题的方法：不要订阅完整的Subject，而只需订阅next-call。但请记住，这只是实现此功能的一种方法，根据您的具体情况，可能会有更好的解决方案可用。

let a = Rx.Observable.from([1,2,3,4,5]).share();
let b = new Rx.Subject();

b.subscribe(value => console.log("New subject value: " + value));
a.subscribe(value => console.log("New observable value: " + value));
a.subscribe(data => b.next(data));

b.next("value1");
b.next("value2");

<script src="https://unpkg.com/rxjs/bundles/Rx.min.js"></script>

Answer 2

这是主题及其内部状态的经典问题。

当您调用 a.subscribe(b) 时，Subject 会收到来自源 Observable a 的所有项目以及将其标记为已停止的 complete 通知。您可以在源代码中看到它：https://github.com/ReactiveX/rxjs/blob/master/src/Subject.ts#L86
然后忽略所有其他信号，您可以在此处看到：https://github.com/ReactiveX/rxjs/blob/master/src/Subject.ts#L56

这意味着这个 Subject 实例将永远不会再发出任何东西。它永远完成了。可以看到complete 信号确实是用do() 操作符发送的：

a.do(val => console.log('do: ' + val), null, () => console.log('do: complete'))
  .subscribe(b)

查看演示：https://jsbin.com/bepupor/1/edit?js,console

因此，在您的情况下，您希望保护主题 b 不会收到 complete 通知。您可以通过仅手动获取和推送next 信号来做到这一点：

a.subscribe(val => b.next(val));

查看演示：https://jsbin.com/bepupor/4/edit?js,console

另一个与您非常相似的问题：Rx.Subject loses events