怎麼理解rxjs？

01-21

小白選手，只知道promise，最近了解到還有rxjs/circle.js。不知道Reactive Extensions這種東西怎麼理解？希望有大神能用通俗的解釋方式，讓我等菜鳥了解下。

術語要用對。你要理解的不是RXJS，而是Observable，RXJS是Observable的Javascript實現。有Promise經驗，理解Observable會容易一些。

var promise = new Promise((resolve, reject) =&> { setTimeout(() =&> resolve("foo"), 1000); });

var observable = Observable.create((observer) =&> { observer.onNext("foo"); observer.onNext("bar"); var complete = setTimeoutout(() =&> observer.onComplete(), 2000); return () =&> { clearTimeout(complete) } });

promise只能resolve一次，輸出單值，observable能輸出多值。promise一旦建立就只有兩種可能，要麼resolve，要麼reject，對應observable的complete，error，但observable可以定義如何cancel。上面的

let disposable = observable.forEach((value) =&> console.log(value)); disposable(); // 在observable輸出任何值前就取消了

Observable是懶惰的，observable對象建立並不立即輸出值，而是等到有人來subscribe的時候才開始輸出。

這裡有講解Promise和Observable的區別的視頻 RxJS Observables vs Promises，如果想了解為什麼Observable有實際用處，可以參考這個網頁 Overview。慢慢看，多看幾遍。

Angular 2的Http庫是一個很實際的例子 Taking advantage of Observables in Angular 2，其他的資料，用google搜索，不是只有中文互聯網才叫互聯網。

直覺上Observable和Promise的區別是功能，Promise的功能相對單一，因為非同步處理單值流要比非同步處理多值流，變化空間要小很多。Observable的變化要多得多 Operators。

正好最近也在看這個，拋開具體的API和Interface通俗的講，你需要將它理解為一個「推」形式的事件流，如果你能把所有處理流程抽象成「推」形式的事件流的話，一切就都迎刃而解了。

既然有「推」形式的，自然也有「拉」形式的。拉形式的比較容易理解，如果你有Linq的經驗，那麼Linq基本可以認為是一個「拉」形式的數據流，先定義數據流的流轉變形轉換投影規則，然後一旦foreach開始使用這個數據流的話，一切就都按照你事前定義的各種規則進行處理了。

「推」形式和「拉」形式的定義方法差不多，大部分也是轉換變形投影等等，只不過定義的時候，數據（事件）可能還未存在，需要由外界稍後提供給你，這一點和「拉」形式的數據源事先就存在有很大區別。（「推」形式的也可以事先有數據）

舉一個具體例子方便思考：

（此例子來源於stackoverflow）

如果我想實現滑鼠按下的時候選擇一個HTML元素，然後滑鼠挪動的時候這個東西跟著我的滑鼠走，滑鼠放開的時候這個元素停止在我滑鼠放開的位置。

這個需求有一定經驗的人都會做，比如設置各種狀態，跟蹤滑鼠的按下，抬起，移動事件，然後不停的修改元素的位置。但是想一次作對不容易，因為各種狀態想保持好也不是非常容易。

如果用Rx的思路考慮呢，可以抽象成這樣：

以滑鼠按下的事件流為起點，轉換為滑鼠移動的事件流，直到滑鼠抬起的事件流為止整個事件流結束。注意這只是定義事件流，並不涉及具體執行。

那麼偽代碼可能是這樣：

// 定義滑鼠抬起的事件流，事件發生在移動對象dragTargets上 var mouseup = Rx.Observable.fromEvent(dragTargets, "mouseup"); // 定義滑鼠移動的事件流，事件發生在document上 var mousemove = Rx.Observable.fromEvent(document, "mousemove"); // 定義滑鼠按下的事件流，事件發生在移動對象dragTargets上 var mousedown = Rx.Observable.fromEvent(dragTargets, "mousedown");


    // 轉換操作，滑鼠按下事件發生的時候，將事件流轉換成跟蹤滑鼠移動事件流

    var mousedrag = mousedown.flatMap(function (md) {
      // 計算移動開始時候元素的位置

      var startX = md.offsetX, startY = md.offsetY;
      // 真正的轉換開始

      return mousemove.map(function (mm) {

        mm.preventDefault();
      // 事件裡面的值，即最後訂閱者可以操作的數據

        return {

          left: mm.clientX - startX,

          top: mm.clientY - startY,

          target: mm.target,

        };

      // 直到滑鼠抬起事件發生的時候整個事件流結束

      }).takeUntil(mouseup);

    });

// 訂閱者的操作，簡單的將元素的style修改即可 var subscription = mousedrag.subscribe(function (d) { d.target.style.top = d.top + "px"; d.target.style.left = d.left + "px"; });

可以發現，上面的例子裡面，稍微緊湊點修改一下就整個只是一個鏈式調用，不存在臨時狀態變數，所有東西都不影響外界，很純粹的函數式寫法。而且，這樣的思考方式也很接近於人類的思考方式。

至於事件流能做什麼操作（operator），那就多得要命了，可以去官方網站學習一下。很多操作是Linq裡面不存在的。

再舉個例子，為了性能考慮，我想讓上面的拖動不要每次滑鼠移動都重繪，想延遲一點點比如500毫秒重新繪製一次怎麼辦？可以用debounceTime這個操作將事件流給控制在500毫秒才給你一次最新值（加在哪裡你可以考慮，是滑鼠移動時候控制，還是最終的事件流控制）整個還是個鏈式操作，很方便調整。

加在最終事件流的例子：

}).debounceTime(500).takeUntil(mouseup);

代替Promise么，就當事件流裡面只有一個事件就好了，這樣理解是不是OK？

要做到通俗易懂，就得弄清楚 RxJS 的一些基礎和概念。

RxJS 整個庫的基礎就是 Observable，注意是 Observable ，不是 Observer，和觀察者模式區別開。

Observable 對象初探

關於對 Observable 對象的理解，通過下面的代碼就能有個大致的印象，看得出來 $clickStream 其內部可以連續產生新事件（或者值），並且在外部可以被監聽。

let button = document.getElementById("btn1"); let clickStream$ = Observable.create(function(observer){ button.addEventListener("click", function(event){ observer.next(event) }, false); }); clickStream$.subscribe(ev =&> console.log(ev));

其中 create 方法接受的那個函數，一般被俺稱做 Producer（生產者）函數，subscribe 方法接受的那個函數，被俺稱作回調函數（噗），然而事實上並不那麼簡單。

Observable 里的 lazy 、cold 和 hot

Rx 中的 Observable 對象默認是 lazy 且 cold 的，lazy 說的是如果 Observable 對象的 subscribe 方法沒有被調用，那麼 Producer 函數也就不會被調用。cold 是說每調用一次 subscribe 方法，就會執行一次 Producer 來產生一個新的 Subscription 流，也就是不同的 Subscription 流內的狀態不會相互影響。

舉個例子（JS Bin）：

let interval = Rx.Observable.create(observer =&> { setInterval(() =&> observer.next(), 1000) }).scan(x =&> x+1, 0); //累加器，從 0 開始，類似在內部定義了一個count，然後每次count++ // 1, 2, 3... interval.subscribe(count =&> console.log("count1:"+count)); setTimeout(() =&> { // 3s 後監聽，同樣得到 1, 2, 3... 而不是 4, 5, 6.. interval.subscribe(count =&> console.log("count2:"+count)); }, 3000);

如果要對象變成 hot 的，也就是不管調用 subscribe 多少次都是監聽同一個流，那麼就調用一下 Observable 對象的 share 方法，把它標記為 hot。上面的例子，如果我們希望在 3s 後監聽是得到4, 5, 6...，那麼在調用完 scan 的地方再調用一下 share 就可以了，這時候可以理解成和第一個subscribe監聽的是同一個「流」。

為啥 Rx 是這麼設計的呢？我覺得也很奇怪。Cycle.js 的作者就對此也感到不解，所以重新設計了一個響應「流」庫：staltz/xstream，只產生 hot 的「流」，我覺得更合理些，只是不知 Observable W3C 規範將來會怎麼定義。

Observable Transform

Observable 對象上除了 subscribe ，還有一堆方法，用來把當前的「流」轉成（transformTo）另一個「流」。比如 map 和 filter ：

--1---3--5-----7------ map(i =&> i * 10) --10--30-50----70-----

--1---2--3-----4-----5---6--7-8-- filter(i =&> i % 2 === 0) ------2--------4---------6----8--

在上面的兩個例子中，函數返回的是新值，有時候還需要支持返回一個流，然後這個流的狀態產生一個新流，比如下面的例子：

---1---2-----3--4----5----6--- endWhen( --------a--b--| ) ---1---2-----3--4--|

Observable 類靜態方法

再說下 Observable 類上的靜態方法，通常是工廠方法，用來定義流的產生（比如上面的 create）；用來將其他對象轉成流，比如 fromPromise，就是將一個 Promise 對象轉成一個「有終點」的流：

fromPromise( ----42 ) -----------------42|

還有用來將兩個流合併，產生一個新的流的方法，比如 merge：

--1----2-----3--------4--- ----a-----b----c---d------ merge --1-a--2--b--3-c---d--4---

舉個實際應用場景：電商 Detail 頁面價格模型

不知不覺說了這麼多，掌握了上面的基本姿勢，我們再來點實踐，比如設計一個電商頁面的價格數據模型層：

在商品詳情頁面上，不同的 sku （最小庫存單元，例如iPhone 7 32GB和 iPhone 7 128GB是兩個不同的sku）價格不同，用戶選擇完具體的一個 sku 之後，要顯示對應的價格，其實可以用流抽象出一個模型層：

首先是選擇 sku 的時候我們產生一個 dom 事件的流：

---ev----ev----ev----ev----

然後把這個事件流轉變成 skuId$ （數據）流：

---skuId----skuId----skuId---skuId----

然後把 skuId$ 流轉成 skuId 對應的價格（數據）流：

---price---price---price--price----

最後我們在 View 層面上 subcribe 一下 price$ 流，在回調函數里修改一下價格標籤的innerHTML：

price$.subscribe(price =&> $el.html(price));

當然，實際場景往往比這複雜多了，電商detail頁面在電商前台頁面中算比較複雜的了。

最後

哈哈，大概就這麼回事了，你可能已經明白了，不明白的話自己再多試幾次。等你覺得自己掌握了，不妨去和別人談笑風生，交流一下各自的理解。

其他的概念：Rx中的subject和scheduler怎麼理解概念？ - Bakso 的回答 - 知乎

感興趣可以跟著系列文章動手試一下 RxJS：

https://zhuanlan.zhihu.com/p/23331432

https://zhuanlan.zhihu.com/p/23464709

直接提供資料....

直接到rx的老家看

https://channel9.msdn.com/Tags/rx

其中,比較系統的介紹是這個DevCamp 2010 Keynote

其中:rxjs最開始的來自這裡 Introducing RxJS

原來coursera還有課程的, Principles of Reactive Programming, 不過現在沒了好像.

---------------------------------------------------------------------------------

看到 @欲三更開始說的

" 電路"啥的,

其實就這樣

參考 Structure and Interpretation of Computer Programs, 2e: 3.3 A Simulator for Digital Circuits

本來 "Principles of Reactive Programming"課程上分3部分+3個老師教的第一個就是用scala實現了上面的離散數字模擬, 第二部分才是Rx

它僅僅做了一件事，把一個範疇的對象和態射提升到另一個範疇。即，

1.提升對象

2.提升態射

在老範疇里，它們邏輯雖然是線性的，但你卻無法用線性的代碼去表達，只能寫分支、回調、各種循環等形式上看起來很髒的代碼。

提升到新的範疇的目的是讓你可以用代碼線性地描述新對象和新態射的關係並且可以複合態射。

它類似於數組和promise。

其中數組是對數據的提升，而它的mapReduce是態射的提升。也就是說對象和態射被提升到了新的範疇。但新範疇的結構與老範疇是同構的。

promise是非同步數據的提升，它的then catch是態射的提升。它抹平了時間的跳躍，對與錯的分支。

它們的用處都是講回調、分支、時間、循環等拍成一條直線，讓你能夠線性地描述態射和對象的關係。

它們本質上都是一種東西，它們都是monad這個概念的不同應用，這個概念能夠保持對象和態射在不同的範疇里的同構現象。

你懂了monad,就懂了數組，promise,Observable,Maybe等概念。

所以你應該先搞懂monad。

RxJS - 推酷

Rx is the underscore.js for events.

可以看看這個教程：

RxJS指南_rxjs教程_匯智網

Rx is the lodash.js for events.

frp = functional reactive program.

functional 就是函數式的思想，通常情況下式用來處理數據的。

reactive 是響應式，cyclejs里有個裡有個很簡單的字。有兩個對象：A和B。

如果A完成某個動作，B需要更新，有兩種方式：

1. A完成後調用B的更新方法；

2. B傾聽A的「完成」事件，然後自己處理更新動作。

兩種方式都能完成B更新的功能，方法1, B是被動的（被A調用的），方法2, B是主動的（自主的，想什麼時候，在哪裡調用都可以）

reactive就是第二種方式，就是觀察者模式，事件機制。

函數式處理的對象由普通數據變成了事件，對事件進行遍歷，過濾，合併等操作。

所以說Rx是事件界對lodash就不難理解了。

說一些不準確但是能夠入門的理解方法吧。

observeable是promise的強化版。

promise是一次性的，observeable是多次的。

有種「對象與類」的感覺，promise討論的是「對於這個非同步事件，要如何處理？」，observeable討論的是「對於這一類非同步事件，要如何處理？」。

關於observeable的各種操作符，實際上不一定要強求。可能會用得到的幾個:

1 想實現類似於promise.then的效果？

switchMap

2 Promise.all？

forkJoin

3 race？

就是race

4 catch？

subscribe時的回調。

你也可以考慮catch操作符，但是記得那需要return一個新的observeable。

上面幾條不一定真的正確。

最後:瞎jb用並不可恥，想不出好辦法時先用toPromise頂著是明智之舉。

那我想提問下這個rxjs和generater又有什麼相似和不相似的地方呢？

generater藉助promise好像也可以實現這種效果

我也是剛剛了解到rxjs，感覺和ES6兩個概念比一下還是很有必要的

promise就是一個非同步調用的對象，在將來某一個時間resolve，並且只能resolve一次。就是那個調用的方法完成以後，要返回的結果用resolve返回。當然你也可以reject，代表返回一個錯誤。

observable更像是一個消息生成器，你可以給他設置一個處理函數，相當於訂閱者。然後你可以通過這個消息生成器，生成一個個的消息，讓處理函數去處理。所以他返回的也是相對於的多個值。

rxjs中的數據其實對應的是stream的概念，promise和stream都是非同步數據的一種表示方法，其中promise對應於單個值，stream則對應於數組。在rxjs上做的filter，map等操作和在數組上的是類似的概念，區別只在於是馬上返回值還是將來某個時間返回值（通過註冊回調函數取得）