filter(cond) 和 flatMap(x => cond? of(x) : EMPTY) 之间有区别吗？

Question

我试图了解这两个 observables 之间的区别。代码的唯一区别是：

/**
* Inside rxjs pipe
*/
if(typeof x === 'number' && x > 3) {
    return of(x);
} else {
    return EMPTY;
}

对比：

.filter(typeof x === 'number' && x > 3)

我正在运行的测试：

const a$ = from([1, 6, '4']).pipe(
            tap(console.log),
            flatMap((x) => {
                if (typeof x === 'number') {
                    if (x > 3) {
                        return of(x);
                    }
                    return EMPTY;
                }
                return EMPTY;
            }),
            tap(console.log)
        );
        const sub_a = a$.subscribe(
            (x) => { console.log(x, 'success'); done(); },
            (e) => { console.log(e, 'error'); done(); },
            () => { console.log('complete'); sub_a.unsubscribe(); done(); }
        );

和：

        const b$ = from([2, 5, '8']).pipe(
            tap(console.log),
            filter(x => typeof x === 'number' && x > 3),
            tap(console.log)
        );
        const sub_b = b$.subscribe(
            (x) => { console.log(x, 'success'); done(); },
            (e) => { console.log(e, 'error'); done(); },
            () => { console.log('complete'); sub_b.unsubscribe(); done(); }
        );

对于他们两个，我将第一个值记录一次（在 filter/flatMap 之前），第二个值从水龙头记录两次，一次“完成”，第三个记录一次。

我认为不同之处在于发出 EMPTY 会导致 observable 完全关闭，但后续值仍然可以通过管道看到。

我对@9​​87654326@ 做了同样的事情，唯一的区别是Subjects 没有发出Complete，这是意料之中的。

Answer 1

如果从flatMap 返回的 Observable 具有不同的调度程序，则可能会有所不同，但在您的示例中，可见行为应该是相同的。如果您依赖副作用，通常会发生这种情况，通常不鼓励这样做。

这是asyncScheduler 改变行为的示例（在第二个示例中创建后打印的值）：

const { of, asyncScheduler, EMPTY, from } = rxjs; // = require("rxjs")
const { filter, flatMap } = rxjs.operators; // = require("rxjs/operators")

const items$ = from([1, 2, 3, 4, 5]);

console.log("------------ SYNC");
const sync$ = items$.pipe(
  filter(v => v % 2 === 0)
);
sync$.subscribe(e => console.log(e));
console.log("after sync");


console.log("------------ ASYNC");
items$.pipe(
  flatMap(v => v % 2 === 0 ? of(v, asyncScheduler) : EMPTY)
).subscribe(e => console.log(e));
console.log("after async");

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