漁人和Rxjs的故事,此次必定教會你前端必會的Rxjs

做者：殷榮檜@騰訊

這篇文章可在個人 github 中查看，若是你以爲寫的還能夠，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的故事就講完了，若是你還沒懂，那就把上面這個故事再看一遍。還沒懂，那就多看幾遍了，哈哈。

你能夠在點擊這裏看一下結果JS Bin

下面是對捕魚的三個階段所碰到問題的解決方案（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的區別詳情能夠參考這篇文章實例解析防抖動（Debouncing）和節流閥（Throttling）

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
// complete

source : --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
// complete

source : ----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
// complete

source : ----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複製代碼

參考資料：

Rxjs官方文檔

30 天精通 RxJS