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

摘要： 作者：殷榮檜@騰訊 這篇文章可在我的 github 中檢視，如果你覺得寫的還可以，Please送上你寶貴的star.寫在最前面：你一定要堅持看完這個故事，看完你一定會懂Rxjs.千萬不要覺得故事情節沒有《盜墓筆記》好看而放棄。因為臣妾實在是隻能把枯燥的程式寫成這個很（挺）有（簡）趣...

作者：殷榮檜@騰訊

這篇文章可在我的 ofollow,noindex">github 中檢視，如果你覺得寫的還可以，Please送上你寶貴的star.

寫在最前面：你一定要堅持看完這個故事，看完你一定會懂Rxjs.千萬不要覺得故事情節沒有《盜墓筆記》好看而放棄。因為臣妾實在是隻能把枯燥的程式寫成這個很（挺）有（簡）趣（陋）的故事了。

故事是這樣的

Rxjs的故事有以上圖中幾個主角，我們來一一介紹，這幾個主角你一定要認識。

(1）Rx.Observable 是一條河流。

(2）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