tl;博士; mergeMap比map. 理解mergeMap是获得 Rx 全部功能的必要条件。
相似之处
差异
地图
合并映射
可以改变源流的大小；对于每个元素，可能会创建/发出任意数量（0、1 或许多）的新元素
它提供了对异步性的完全控制——无论是何时创建/发出新元素，以及应同时处理多少来自源流的元素；例如，假设源流发出 10 个元素，但maxConcurrency设置为 2，则将立即处理两个第一个元素，其余 8 个元素将被缓冲；一旦其中一个被处理complete，源流中的下一个元素将被处理等等 - 这有点棘手，但请看下面的例子
所有其他操作符都可以只用mergeMap和Observable构造函数来实现
可用于递归异步操作
返回值必须是 Observable 类型（或者 Rx 必须知道如何从中创建可观察的 - 例如Promise，数组）
id： x => Rx.Observable.of(x)
数组类比
let array = [1,2,3]
fn map mergeMap
x => x*x [1,4,9] error /*expects array as return value*/
x => [x,x*x] [[1,1],[2,4],[3,9]] [1,1,2,4,3,9]
类比不显示完整图像，它基本上对应于.mergeMap与maxConcurrency设置为1。在这种情况下的元素将被排序为以上，但在一般情况下，它不必须是如此。我们唯一的保证是新元素的发射将按照它们在底层流中的位置进行排序。例如：[3,1,2,4,9,1]and[2,3,1,1,9,4]是有效的，但[1,1,4,2,3,9]不是（因为4是2在底层流中发出的）。
几个使用的例子mergeMap：
// implement .map with .mergeMap
Rx.Observable.prototype.mapWithMergeMap = function(mapFn) {
return this.mergeMap(x => Rx.Observable.of(mapFn(x)));
}
Rx.Observable.range(1, 3)
.mapWithMergeMap(x => x * x)
.subscribe(x => console.log('mapWithMergeMap', x))
// implement .filter with .mergeMap
Rx.Observable.prototype.filterWithMergeMap = function(filterFn) {
return this.mergeMap(x =>
filterFn(x) ?
Rx.Observable.of(x) :
Rx.Observable.empty()); // return no element
}
Rx.Observable.range(1, 3)
.filterWithMergeMap(x => x === 3)
.subscribe(x => console.log('filterWithMergeMap', x))
// implement .delay with .mergeMap
Rx.Observable.prototype.delayWithMergeMap = function(delayMs) {
return this.mergeMap(x =>
Rx.Observable.create(obs => {
// setTimeout is naive - one should use scheduler instead
const token = setTimeout(() => {
obs.next(x);
obs.complete();
}, delayMs)
return () => clearTimeout(token);
}))
}
Rx.Observable.range(1, 3)
.delayWithMergeMap(500)
.take(2)
.subscribe(x => console.log('delayWithMergeMap', x))
// recursive count
const count = (from, to, interval) => {
if (from > to) return Rx.Observable.empty();
return Rx.Observable.timer(interval)
.mergeMap(() =>
count(from + 1, to, interval)
.startWith(from))
}
count(1, 3, 1000).subscribe(x => console.log('count', x))
// just an example of bit different implementation with no returns
const countMoreRxWay = (from, to, interval) =>
Rx.Observable.if(
() => from > to,
Rx.Observable.empty(),
Rx.Observable.timer(interval)
.mergeMap(() => countMoreRxWay(from + 1, to, interval)
.startWith(from)))
const maxConcurrencyExample = () =>
Rx.Observable.range(1,7)
.do(x => console.log('emitted', x))
.mergeMap(x => Rx.Observable.timer(1000).mapTo(x), 2)
.do(x => console.log('processed', x))
.subscribe()
setTimeout(maxConcurrencyExample, 3100)
<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.1.1/Rx.min.js"></script>