RxJs如何将两个重叠的可观察到的合并成一个

Question

我有两个可观察：

-1-2-3-4-5-6-7-8-9-10-11-12-13-14-15-| 
-13--14--15--16--17--18--19-----20---------21--------------22------23--24--> 

第一个包含一些越来越多，但经过一段时间后停止（这些是来自数据库的光标结果） 第二类是不断涌现越来越多的人首先包含一些数字，但不要停止发射。 （这些都是新插入的数据到数据库）

我想这两个可观察一下一个连续的观察到这样的：

-1-2-3-4-5-6-7-8-9-10-11-12-13-14-15-16-17-18-19-20-21-----22------23--24--> 

此观察到的包含了每一个号码只有一次，保持顺序发光。

如何使用尽可能少的内存来解决问题？

Answer 1

我认为最好的办法在这里是缓冲b $，直到$流达到b $，然后发出所有b $的缓冲项并切换到b $。事情是这样的：

const a = '-1-2-3-4-5-6-7-8-9-10-11-12-13-14-15'; 
 
const b = '-13--14--15--16--17--18--19-----20---------21--------------22------23--24'; 
 

 
const fromMarble = str => Rx.Observable.from(str.split('-')).concatMap(x => Rx.Observable.of(x).delay(1)).filter(v => v.length).map(x => parseInt(x)); 
 

 
const a$ = fromMarble(a).share(); 
 
const b$ = fromMarble(b).share(); 
 

 
const switchingSignal$ = Rx.Observable.combineLatest(a$, b$.take(1), (a, b) => a >= b).filter(x => x).take(1).share(); 
 

 
const distinct$ = Rx.Observable.merge(
 
\t a$.takeUntil(switchingSignal$).map(x => x + '(from a)'), 
 
\t b$.buffer(switchingSignal$).take(1).mergeMap(buffered => Rx.Observable.from(buffered)).map(x => x + '(from b$ buffer)'), 
 
\t b$.skipUntil(switchingSignal$).map(x => x + '(from b$)') 
 
); 
 

 
distinct$.subscribe(console.log);

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.5.0/Rx.js"></script>

Answer 2

您可以通过采取从（.concat）与第二流连接起来的第一流的所有元素，除了最新的一个（.last）之前（.skipWhile含）的元素

const a = '-1-2-3-4-5-6-7-8-9-10-11-12-13-14-15' 
 
const b = '-13--14--15--16--17--18--19-----20---------21--------------22------23--24' 
 
const fromMarble = str => Rx.Observable.defer(() => { 
 
    console.log('side effect from subscribing to: ' + str); 
 
    return Rx.Observable.from(str.split('-').filter(v => v.length)); 
 
}); 
 

 
const a$ = fromMarble(a); 
 
const b$ = fromMarble(b); 
 

 
const distinct$ = Rx.Observable.concat(
 
    a$, 
 
    a$.last().switchMap(latest => 
 
    // .skipWhile + .skip(1) => skipWhile but inclusive 
 
    b$.skipWhile(v => v !== latest).skip(1) 
 
), 
 
); 
 

 
distinct$.subscribe(e => console.log(e));

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

做到这一点另外，如果订阅时有副作用（例如订阅时 - 将创建新光标），则可以通过使用例如const a$ = fromMarble(a).shareReaplay()来为所有订户共享该副作用。

你可以阅读更多有关共享的副作用：老文档中

• 为RxJS V4 - 4.8 Use the publish operator to share side-effects
• 并从这篇文章 - RxSwift: share vs replay vs shareReplay