渔人和Rxjs的故事,这次一定教会你前端必会的Rxjs-白红宇

作者：殷荣桧@腾讯

这篇文章可在我的中查看，如果你觉得写的还可以，Please送上你宝贵的star.

写在最前面：你一定要坚持看完这个故事，看完你一定会懂Rxjs.千万不要觉得故事情节没有《盗墓笔记》好看而放弃。因为臣妾实在是只能把枯燥的程序写成这个很（挺）有（简）趣（陋）的故事了。

故事是这样的

Rxjs的故事有以上图中几个主角，我们来一一介绍，这几个主角你一定要认识。

(1）Rx.Observable 是一条河流。是人们赖以生活的一个环境。以此为基础，才有接下来围绕这条河流谋生的一个群体。

(2）source 作为一条在河流中捕鱼船上的竹筒（相当于把从服务器获取的数据都塞进这条筒中，形成数据流source）。鱼（data）可以一个一个的钻到竹筒中（source)

var source = Rx.Observable.create(subscriber) 复制代码

(3) subscriber 是位捕鱼的渔人，是位好心人,主要任务是把捕获的鱼（data)扔向岸边的饥民。其实渔人就是拿到服务器端数据后如何做分发的管理者。

var subscriber = function(observer) {    var fishes = fetch('http://www.oa.com/api'); // 捕获到鱼    observer.next(fishes.fish1); // 把捕获的第一条鱼扔向岸边的饥民    observer.next(fishes.fish2); // 把捕获的第二条鱼扔向岸边的饥民}复制代码

(4)observer 作为岸边上饥民。其实就是从服务器端获取数据后的最终消费者，他们决定怎么用这些数据展示到用户的页面上，就和饥民拿到鱼后决定怎么烹饪是一个道理。因为来自天南地北，方言不同，所以描述自己在获取到鱼后的吃法表述时语法不同，但其实实质都是一样的，有鱼了（value=> {})怎么办,没鱼了（error => {})怎么办，当天的鱼扔完了（complete => {}）怎么办。

方式一：

observer = (value => { console.log(value); },    error => { console.log('Error: ', error); },    () => { console.log('complete') })source.subscribe(observer)复制代码

方式二：

observer = function(value) {    console.log(value);}source.subscribe(observer); // 这根捕鱼的竹筒很多饥民都翘首以待（subscribe),所以竹筒（source)会被新来的饥民订阅（subscribe).当然，饥民不订阅自然渔人就不会把竹筒(source)中捕获的鱼扔给他。复制代码

方式三：

observer = {    next: function(value) {        console.log(value);    },    error: function(error) {        console.log('Error: ', error)    },    complete: function() {        console.log('complete')    }}source.subscribe(observer);subscribe 河流source知道河流的两边有哪些百姓需要救济，所以会帮助他subscribe渔人扔出的鱼，这样他就会收到鱼了source.subscribe(observer);复制代码

(5)subscription 为哪个饥民订阅了哪个竹筒的清单。可以从清单上划去，那么这个饥民就再不会受到渔人扔出的鱼了

subscription = source.subscribe(observer1);subscription.unsubscribe(); // 从清单上划去饥民observer1的订阅信息，因为observer1已经不是饥民了，不需要救济了。复制代码

我们把上述的五个角色链接起来就是rxjs的实现过程，我们先用易懂的拼音试一下，再对应到真正的rxjs语法。

var 渔人 = function (饥民) {    var fishes = fetch('server/api'); // 捕获到一定数量的鱼    饥民.next(fishes.fish1); // 接下来把鱼1扔给饥民    饥民.next(fishes.fish1); // 接下来把鱼1扔给饥民} var 饥民1 = { // 饥民要想好不同种情况下的应对方法，不能在没有捕到鱼的时候就饿死。    next:function (fish) {        // 有鱼扔过来了，把fish煮了吃掉。    }，    error: function(error) {       // 捕获的鱼有毒，不能吃，所以要想其他办法填饱肚子，可以选择吃野菜什么的，    }，    complete: function() {        // 当天的鱼扔完了，那么可以回家了    }}var 竹筒 = 河流.create(渔人)； // 河流中来了一名渔人，那么他一定会在河流中放下捕鱼的竹筒。清单 = 竹筒.subscribe(饥民1) // 竹筒被饥民1关注后，就可以收到渔人扔出的鱼了。setTimeout(() => {        清单.unsubscribe();  // 一年后，饥民摆脱困境，不再需要救济，就退订这个竹筒了。把机会让给别人。}, 1年);复制代码

对应到真正的rxjs语法，我们再来一遍。

var subscriber = function(observer) { // 创建了一位渔人    observer.next('fish1');    observer.next('fish2');    observer.complete();}var observer1 = { // 来了一位饥民1    next: function(value) {        console.log(`我接到鱼${value}啦，不会挨饿咯`);    },    error: function(error) {        console.log(`哎，捕到的鱼因为${error}原因不能吃`)    },    complete: function() {        console.log('今天的鱼发完了')    }}var source = Rx.Observable.create(subscriber); // 河流中来了一名渔人，他在河流中放下捕鱼的竹筒。subscription = source.subscribe(observer1); // 竹筒被饥民1关注后，饥民1可以收到渔人扔出的鱼了。setTimeout(()=> {    subscription.unsubscribe(); // 3秒后饥民退订了竹筒，给其他饥民机会。}, 3000);打印出的结果如下：// "我接到鱼fish1唠"// "我接到鱼fish2唠"// "今天的鱼发完了"复制代码

到此为止Rxjs的故事就讲完了，如果你还没懂，那就把上面这个故事再看一遍。还没懂，那就多看几遍了，哈哈。

你可以在点击这里看一下结果

下面是对捕鱼的三个阶段所碰到问题的解决方案（1）竹筒中如何才能产生鱼（2）竹筒中有鱼了，怎么向外取（3）取出来后，鱼被扔向岸边的过程中发生了什么。所以操作符的使用也是有先后顺序的。

一.竹筒中如何才能产生鱼

(1) create 在事先没有鱼的情况下，使用create从水下fetch

var source = Rx.Observable    .create(function(observer) {          var fishes = waitForFishes_ajax_fetch(api);        observer.next(fish.fish1);        observer.next(fish.fish2);        observer.complete();    });复制代码

(2) of(arg1,arg2)

当鱼是现成的，但是是散装的时候，比如昨天还存了几条在船上，用of装到竹筒中

var source = Observable.of(fish1,fish2);复制代码

(3)from ([arg1,arg2,arg3]);

当于是现成的，同时用草绳穿成一排时(为数组结构)，需要用from方法装到竹筒中

var fishes = [fish1, fish2];var source = Observable.from(fishes);注：from 还能够传入字符串var source = Rx.Observable.from('铁人赛');// 铁// 人// 赛// complete!复制代码

(4)fromEvent(document.body,'click');

除了向岸上扔鱼以外，有时候河里发生的事件（船体（document.body）被浪击打（click)）的内容（target.event)渔人也会用竹筒作为喇叭告诉岸上的饥民，让他们做好今天情况不太好的准备。

var source = Rx.Observable.fromEvent(document.body, 'click');复制代码

(5) empty,never,throw

var source = Rx.Observable.empty(); // 一条鱼都没有捕捉到的情况,直接触发observer中complete的执行结果为 // complete!var source = Rx.Observable.never();  // 渔人累了，不管是捕到鱼还是捕不到鱼都没有力气向岸边上的饥民发出告知了。结果为 // complete永远都不会触发var source = Rx.Observable.throw('ill'); // 当渔人生病了，或者要去会个老朋友，会向岸边的饥民（observer)用竹筒呐喊一声告知，这样饥民就想别的办法（触发error方法）解决当天的食物问题。复制代码

（6） interval('间隔时间')

Rx.Observable.interval(1000) // 渔人每天捕鱼也很无聊，想和岸上的饥民搞个游戏，每过1秒钟向岸上的饥民扔一条鱼(而且还在鱼身上表上0，1，2，3....)，并且让饥民拿到鱼之后，只要鱼上的数字timer('第一条鱼的扔出等待时间'，‘第一条之后扔鱼的间隔’) Rx.Observable.timer(1000, 5000); // 游戏规则改了一点，渔人告诉饥民，他会在1000毫秒之后才会向岸边扔出第一条鱼，以后每隔5000毫秒扔出一条。复制代码

二.竹筒中有鱼了，怎么向外取

2.1 单个竹筒捕鱼

（1） take

渔人决定只取竹筒中的前三条，因为怕竭泽而渔。

var source = Rx.Observable.interval(1000);var example = source.take(3);example.subscribe({    next: (value) => { console.log(value); },    error: (err) => { console.log('Error: ' + err); },    complete: () => { console.log('complete'); }});// 0// 1// 2// complete复制代码

（2） first

first 同take(1)是一个意思，表示只取第一条鱼

var source = Rx.Observable.interval(1000);var example = source.first();example.subscribe({    next: (value) => { console.log(value); },    error: (err) => { console.log('Error: ' + err); },    complete: () => { console.log('complete'); }});// 0// complete复制代码

(3) takeUntil

takeUntil 是当渔人从竹筒中取鱼时，当遇到一条特殊的鱼（比如遇到一条金色的金龙鱼）之后，就不会再取了。因为再取就不太吉利，就会得罪龙王了（参照《西游记》第XX篇）。

(4) concatAll()

把两竹筒的鱼串联合并成一竹筒的鱼然后取出。

(5) skip

var source = Rx.Observable.interval(1000);var example = source.skip(3); // 忽略竹筒中的前几条鱼，然后取后面的鱼example.subscribe({    next: (value) => { console.log(value); },    error: (err) => { console.log('Error: ' + err); },    complete: () => { console.log('complete'); }});// 3// 4// 5...复制代码

（6）takeLast（）

var source = Rx.Observable.interval(1000).take(6);var example = source.takeLast(2); // 表示只取竹筒中的最后两条鱼example.subscribe({    next: (value) => { console.log(value); },    error: (err) => { console.log('Error: ' + err); },    complete: () => { console.log('complete'); }});// 4// 5// complete复制代码

（7） last()

var source = Rx.Observable.interval(1000).take(6);var example = source.last(); // 相当于就是takeLast(1),表示只取竹筒中最后一条鱼example.subscribe({    next: (value) => { console.log(value); },    error: (err) => { console.log('Error: ' + err); },    complete: () => { console.log('complete'); }});// 5// complete复制代码

(8) concat(observable1,observable2,....)

同样是把所有的竹筒串起来，然后把鱼取出来

var source = Rx.Observable.interval(1000).take(3);var source2 = Rx.Observable.of(3)var source3 = Rx.Observable.of(4,5,6)var example = source.concat(source2, source3);  // 与concatAll()不同的concatAll([observale1,observable2...])中是数组，而concat(observable1,observable2,....)中是一个一个的参数example.subscribe({    next: (value) => { console.log(value); },    error: (err) => { console.log('Error: ' + err); },    complete: () => { console.log('complete'); }});// 0// 1// 2// 3// 4// 5// 6// complete复制代码

（9） startWith()

可能当天捕到的鱼不是很多，不够岸边的饥民吃。渔人就偷偷在竹筒前面塞几条进去，假装今天捕到了很多鱼，然后取出。

var source = Rx.Observable.interval(1000);var example = source.startWith(0); // 渔人变了一条鱼塞在前面example.subscribe({    next: (value) => { console.log(value); },    error: (err) => { console.log('Error: ' + err); },    complete: () => { console.log('complete'); }});// 0// 0// 1// 2// 3...复制代码

（10）scan

当需要对所有的捕捉到的鱼做一个统计时，比如统计所有鱼的总重量，就需要扫描（scan)每一条鱼称重，并且用上一条的重量加上下一条的重量，如此累计。

var source = Rx.Observable.from('hello')             .zip(Rx.Observable.interval(600), (x, y) => x);var example = source.scan((origin, next) => origin + next, '');example.subscribe({    next: (value) => { console.log(value); },    error: (err) => { console.log('Error: ' + err); },    complete: () => { console.log('complete'); }});// h// he// hel// hell// hello// complete复制代码

（11） buffer,bufferCount,bufferTime

渔人觉得每捕到一条鱼就扔向岸边太累了，他决定每过一定的时间攒够了一定数量的鱼再取出（bufferCount(3)),或者每过一段时间（bufferTime(1000)）再取出筒中的鱼.或者他甚至可以看到每当第二个筒子中捕满5条鱼时var example = source.buffer(source2); ，就取出所有鱼向岸边扔出。

var source = Rx.Observable.interval(300);var source2 = Rx.Observable.interval(1000);var example = source.buffer(source2);var example = source.bufferTime(1000);var example = source.bufferCount(3);example.subscribe({    next: (value) => { console.log(value); },    error: (err) => { console.log('Error: ' + err); },    complete: () => { console.log('complete'); }});复制代码

（12） delay()

当捕获到一串鱼后，渔人决定抽一支烟后再开始取出鱼

var source = Rx.Observable.interval(300).take(5);var example = source.delay(500); // 渔人用500毫秒的时间抽完烟后再开始扔鱼source : --0--1--2--3--4|        delay(500)example: -------0--1--2--3--4|delayWhen('一定条件')delayWhen((x) => {
    if(x==3) {
    return Rx.Observable.empty().delay(500)}}) // 当扔到第三条鱼时，渔人决定停下来用500毫秒抽支烟再继续扔复制代码

(13) debounceTime

有时候捕鱼，鱼上钩太快，渔人年纪大，来不及一条一条的取。所以他决定鱼高频上钩时不取出向岸上扔（来不及啊），等有两条鱼上钩的时间间隔够大时，能缓够劲来。再一次性把之前的都取出。两次鱼捕获的时间间隔要大于debounceTime，才将上一批次捕获的鱼取出，扔向岸边。

--1--2--3---------5--  // 3,5之间大于debounceTime了，一次取出1，2，3扔向岸边复制代码

(14) throttle

在（13）中有时捕鱼间隔时间长，有时捕鱼间隔时间短，渔人可以在间隔长的时间休息后把上一批攒下的鱼取出。但是当到了夏季捕鱼季时，上钩的鱼根本停不下来，渔人没法采用debounce策略得到休息时怎么办呢（来一条仍一条，渔人会累死），所以渔人又想了一个办法，每过 5秒（throttleTime(5000)）取一条刚好上钩的鱼扔出，或者这会没有鱼上钩就等到一会儿有鱼上钩为止，扔出去之后再等5秒，如此循环，其他时间上钩的鱼就不管了，反正鱼多，够吃。

注：对于debounce与throttle的区别详情可以参考这篇文章

var source = Rx.Observable.interval(300).take(20);var example = source.throttleTime(1000);example.subscribe({    next: (value) => { console.log(value); },    error: (err) => { console.log('Error: ' + err); },    complete: () => { console.log('complete'); }});// 0// 4// 8// 12// 16// "complete"复制代码

(15) distinct

逢年过节，渔人想给百姓来点独一无二的，每次取出鱼时只取不同种类的鱼，让他们好过把吃日本料理的瘾。渔人只取出品种不同的鱼，之前出现过的鱼都抛弃掉。

var source = Rx.Observable.from(['a', 'b', 'c', 'a', 'b'])            .zip(Rx.Observable.interval(300), (x, y) => x);var example = source.distinct()example.subscribe({    next: (value) => { console.log(value); },    error: (err) => { console.log('Error: ' + err); },    complete: () => { console.log('complete'); }});// a// b// c// completesource : --a--b--c--a--b|            distinct()example: --a--b--c------|复制代码

2.2多竹筒捕鱼，鱼怎么向外取

多流的存在，例如下面这些

var source = Rx.Observable.fromEvent(document.body, 'click');var example = source                .map(e => Rx.Observable.interval(1000).take(3))  // 到这一步了才应该考虑到多竹筒捕鱼操作，在这之前，都不需要考虑多竹筒捕鱼操作符的存在。                .concatAll();复制代码

(1) concatAll()

当有多个竹筒捕鱼时，把捕获到鱼的竹筒，一个一个的串联起来，然后取出鱼。

var click = Rx.Observable.fromEvent(document.body, 'click');var source = click.map(e => Rx.Observable.interval(1000).take(3));var example = source.concatAll();example.subscribe({    next: (value) => { console.log(value); },    error: (err) => { console.log('Error: ' + err); },    complete: () => { console.log('complete'); }});// 0// 1// 2// 0// 1// 2复制代码

（2） zip

（两个竹筒中，都是第一条上钩的鱼绑一块取出，都是第二条上钩的鱼绑一块取出）

var source = Rx.Observable.interval(500).take(3);var newest = Rx.Observable.interval(300).take(6);var example = source.zip(newest, (x, y) => x + y);example.subscribe({    next: (value) => { console.log(value); },    error: (err) => { console.log('Error: ' + err); },    complete: () => { console.log('complete'); }});// 0// 2// 4// completesource : ----0----1----2|newest : --0--1--2--3--4--5|    zip(newest, (x, y) => x + y)example: ----0----2----4|复制代码

（3）switch

switch本身就是切换的意思，那这就很好理解了。当a,b,c三个竹筒在捕鱼上，a捕获到鱼了，渔人就一直盯着a筒取鱼，直到一会儿其他筒有鱼捕获时。当一会儿b筒中有鱼捕获时，渔人就切换（switch）视线一直盯着b筒，让后一直从b筒中取鱼，直到其他筒有鱼捕获。如此循环。

var click = Rx.Observable.fromEvent(document.body, 'click');var source = click.map(e => Rx.Observable.interval(1000));var example = source.switch();example.subscribe({    next: (value) => { console.log(value); },    error: (err) => { console.log('Error: ' + err); },    complete: () => { console.log('complete'); }});click  : ---------c-c------------------c--..         map(e => Rx.Observable.interval(1000))source : ---------o-o------------------o--..                   \ \                  \----0----1--...                    \ ----0----1----2----3----4--...                     ----0----1----2----3----4--...                     switch()example: -----------------0----1----2--------0----1--...复制代码

（4) merge(observable2)

分分钟注视着两个竹筒，一个有了取一个，两个同时有鱼了，就同时把两个筒子中的鱼取出。

var source = Rx.Observable.interval(500).take(3);var source2 = Rx.Observable.interval(300).take(6);var example = source.merge(source2);example.subscribe({    next: (value) => { console.log(value); },    error: (err) => { console.log('Error: ' + err); },    complete: () => { console.log('complete'); }});// 0// 0// 1// 2// 1// 3// 2// 4// 5// complete复制代码

（5）mergeAll

在上面的（4）中提到了merge的用法，merge是渔人分分钟注视着两个竹筒，一个有了取一个，两个同时有鱼了，就同时把鱼取出。而mergeAll是渔人分分钟同时注视着多个竹筒，一个有了取一个，两个同时有鱼了，就同时取出两个筒中的鱼，多个同时有了，就一把同时都取出。

var click = Rx.Observable.fromEvent(document.body, 'click');var source = click.map(e => Rx.Observable.interval(1000));var example = source.mergeAll();example.subscribe({    next: (value) => { console.log(value); },    error: (err) => { console.log('Error: ' + err); },    complete: () => { console.log('complete'); }});click  : ---------c-c------------------c--..         map(e => Rx.Observable.interval(1000))source : ---------o-o------------------o--..                   \ \                  \----0----1--...                    \ ----0----1----2----3----4--...                     ----0----1----2----3----4--...                     switch()example: ----------------00---11---22---33---(04)4--...复制代码

(6) combineLatest()

把两个竹筒中最新出现的鱼，取出

var source = Rx.Observable.interval(500).take(3);var newest = Rx.Observable.interval(300).take(6);var example = source.combineLatest(newest, (x, y) => x + y);example.subscribe({    next: (value) => { console.log(value); },    error: (err) => { console.log('Error: ' + err); },    complete: () => { console.log('complete'); }});// 0// 1// 2// 3// 4// 5// 6// 7// completesource : ----0----1----2|newest : --0--1--2--3--4--5|    combineLatest(newest, (x, y) => x + y);example: ----01--23-4--(56)--7|复制代码

2.3附：多竹筒捕鱼快捷操作

从上述多竹筒捕鱼操作可以看出，当采用多竹筒捕获鱼时，往往concatAll,switch,mergeAll这些多竹筒操作符都需要和map操作符结合起来使用，于是，渔人就决定用第一个操作符直接替代这两个操作符，加快取鱼的操作。具体如下：

（1）concatMap

var source = Rx.Observable.fromEvent(document.body, 'click');var example = source                .map(e => Rx.Observable.interval(1000).take(3))                .concatAll();简化如下：var source = Rx.Observable.fromEvent(document.body, 'click');var example = source                .concatMap(                    e => Rx.Observable.interval(100).take(3)                );复制代码

（2）switchMap

var source = Rx.Observable.fromEvent(document.body, 'click');var example = source                .map(e => Rx.Observable.interval(1000).take(3))                .switch();简化如下：var source = Rx.Observable.fromEvent(document.body, 'click');var example = source                .switchMap(                    e => Rx.Observable.interval(100).take(3)                );复制代码

（3）mergeMap

var source = Rx.Observable.fromEvent(document.body, 'click');var example = source                .map(e => Rx.Observable.interval(1000).take(3))                .mergeAll();简化如下：var source = Rx.Observable.fromEvent(document.body, 'click');var example = source                .mergeMap(                    e => Rx.Observable.interval(100).take(3)                );复制代码

（三）取出来后，鱼被扔向岸边的过程中发生了什么

(1)map(callback)

var source = Rx.Observable.interval(1000); var newest = source.map(x => x + 1)  // 当渔人扔出一条鱼后，在鱼飞向岸变得过程中，经过了map射线照射区域,发生变异，体重自动增加了一斤，饥民拿到鱼的时候也就比渔人扔出的要重一斤多。newest.subscribe(console.log);结果为：// 1// 2// 3// 4// 5..复制代码

（2） mapTo()

var source = Rx.Observable.interval(1000);var newest = source.mapTo(2);  // 当渔人扔出一条鱼后，在鱼飞向岸变得过程中，经过了mapTo射线照射区域,发生变异，体重无论胖瘦全部都变为2，饥民拿到鱼就都是2斤重的了。newest.subscribe(console.log);// 2// 2// 2// 2..复制代码

（3） filter()

var source = Rx.Observable.interval(1000);var newest = source.filter(x => x % 2 === 0);  // 当渔人扔出一条鱼后，在鱼飞向岸变得过程中，经过了filter射线照射区域,filter射线就像一堵墙一样，挡住体重不符合标准的鱼，饥民拿到的鱼就个个头很大的鱼。newest.subscribe(console.log);// 0// 2// 4// 6..复制代码

（4) catch()

Fish被扔出，在天空中飞行被操作符变异时，发生意外（比如变异死了，变异焦了）。岸上的百姓要有应急的预案，要么吃野果，或者...不能变异出问题了，岸上的饥民就饿死。

var source = Rx.Observable.from(['a','b','c','d',2])            .zip(Rx.Observable.interval(500), (x,y) => x);var example = source                .map(x => x.toUpperCase())                .catch(error => Rx.Observable.of('h')); example.subscribe({    next: (value) => { console.log(value); },    error: (err) => { console.log('Error: ' + err); },    complete: () => { console.log('complete'); }}); source : ----a----b----c----d----2|        map(x => x.toUpperCase())         ----a----b----c----d----X|        catch(error => Rx.Observable.of('h'))example: ----a----b----c----d----h|  复制代码

（5） retry（）

当fish被扔出，经过天空中的变异操作符时，当该变异过程很有可能失败（比如鱼的体重变异成两倍），可以使用retry（）再让渔人再扔一次。当然还可以规定retry(5)五次（可自定义retry次数）；

var source = Rx.Observable.from(['a','b','c','d',2])            .zip(Rx.Observable.interval(500), (x,y) => x);var example = source                .map(x => x.toUpperCase())                .retry();example.subscribe({    next: (value) => { console.log(value); },    error: (err) => { console.log('Error: ' + err); },    complete: () => { console.log('complete'); }}); 复制代码

（6） repeat

同retry一样，retry是在天空中变异出错时，让渔人重新扔一次。如果变异成功了，说明实验成功（鱼成功在空中由1斤变异为2斤），同样也可以让渔人再来一条。但这时候就要用repeat告诉渔人再来一条了，而不是retry,不然渔人还以为刚才的变异实验没成功呢。

var source = Rx.Observable.from(['a','b','c'])            .zip(Rx.Observable.interval(500), (x,y) => x);var example = source.repeat(1);example.subscribe({    next: (value) => { console.log(value); },    error: (err) => { console.log('Error: ' + err); },    complete: () => { console.log('complete'); }});// a// b// c// a// b// c// complete复制代码

参考资料：