摘要：形式上比普通函數(shù)直接返回值啰嗦一些。這里和的重要區(qū)別在于，模式下，可以決定什么時(shí)候返回值，以及返回幾個(gè)值即調(diào)用回調(diào)函數(shù)的次數(shù)。

RxJS 的 Observable 有點(diǎn)難理解，其實(shí) RxJS 相關(guān)的概念都有點(diǎn)難理解。畢竟 RxJS 引入了響應(yīng)式編程這種新的模式，會(huì)不習(xí)慣是正常的。不過(guò)總得去理解嘛，而認(rèn)識(shí)新的事物時(shí)，如果能夠參照一個(gè)合適的已知事物比對(duì)著，會(huì)比較容易理解吧。對(duì)于 Observable，類比 JS 中的函數(shù)，還是比較好的。

先來(lái)看一個(gè)普通函數(shù)調(diào)用的例子：

function foo() {
  console.log("process...")
}

foo()
// 輸出：
// process...

很簡(jiǎn)單，函數(shù) foo() 封裝了一段邏輯（這里只是向控制臺(tái)輸出），然后通過(guò)調(diào)用函數(shù)，函數(shù)執(zhí)行內(nèi)部的邏輯。

再來(lái)看 RxJS Observable 的一個(gè)例子：

var foo = Rx.Observable.create(() => {
  console.log("process...")
})

foo.subscribe()
// 輸出：
// process...

上例中，通過(guò) Rx.Observable.create() 來(lái)創(chuàng)建 Observable 對(duì)象，將同樣將一段代碼邏輯封裝到 Observable 對(duì)象 foo 中，然后通過(guò) foo.subscribe() 來(lái)執(zhí)行封裝的代碼邏輯。

對(duì)于普通函數(shù)和 Observable 對(duì)象，封裝的代碼邏輯在每次調(diào)用時(shí)都會(huì)重新執(zhí)行一次。從這一點(diǎn)來(lái)看，Observable 能夠和普通函數(shù)一樣實(shí)現(xiàn)封裝代碼進(jìn)行復(fù)用。

函數(shù)調(diào)用后可以有返回值：

function foo() {
  console.log("process...")
  return 42
}

console.log(foo())
// 輸出：
// process...
// 42

Observable 執(zhí)行后也會(huì)產(chǎn)生值，不過(guò)和函數(shù)直接返回的方式不同，要通過(guò)回調(diào)函數(shù)方式獲取：

var foo = Rx.Observable.create((observer) => {
  console.log("process...")
  observer.next(42)
})

foo.subscribe(value => console.log(value))
// 輸出：
// process...
// 42

Observable 對(duì)象內(nèi)部是通過(guò) observer.next(42) 這種方式返回值，而調(diào)用方則通過(guò)回調(diào)函數(shù)來(lái)接收返回的數(shù)據(jù)。形式上比普通函數(shù)直接返回值啰嗦一些。

從調(diào)用方的角度來(lái)看，兩個(gè)過(guò)程分別是：

普通函數(shù)：調(diào)用 > 執(zhí)行邏輯 > 返回?cái)?shù)據(jù)

Observable：訂閱（subscribe） > 執(zhí)行邏輯 > 返回?cái)?shù)據(jù)

從獲取返回值方式來(lái)看，調(diào)用函數(shù)是一種直接獲取數(shù)據(jù)的模式，從函數(shù)那里“拿”（pull）數(shù)據(jù)；而 Observable 訂閱后，是要由 Observable 通過(guò)間接調(diào)用回調(diào)函數(shù)的方式，將數(shù)據(jù)“推”（push）給調(diào)用方。

這里 pull 和 push 的重要區(qū)別在于，push 模式下，Observable 可以決定什么時(shí)候返回值，以及返回幾個(gè)值（即調(diào)用回調(diào)函數(shù)的次數(shù)）。

var foo = Rx.Observable.create((observer) => {
  console.log("process...")
  observer.next(1)
  setTimeout(() => observer.next(2), 1000)
})

console.log("before")
foo.subscribe(value => console.log(value))
console.log("after")
// 輸出：
// before
// process...
// 1
// after
// 2

上面例子中，Observable 返回了兩個(gè)值，第1個(gè)值同步返回，第2個(gè)值則是過(guò)了1秒后異步返回。

也就是說(shuō)，從返回值來(lái)說(shuō)，Observable 相比普通函數(shù)區(qū)別在于：

可以返回多個(gè)值

可以異步返回值

函數(shù)執(zhí)行可能出現(xiàn)異常情況，例如：

function foo() {
  console.log("process...")
  throw new Error("BUG!")
}

我們可以捕獲到異常狀態(tài)進(jìn)行處理：

try {
   foo()
} catch(e) {
  console.log("error: " + e)
}

對(duì)于 Observable，也有錯(cuò)誤處理的機(jī)制：

var foo = Rx.Observable.create((observer) => {
  console.log("process...")
  observer.error(new Error("BUG!"))
})

foo.subscribe(
  value => console.log(value),
  e => console.log("error: " + e)
)

Observable 的 subscribe() 方法支持傳入額外的回調(diào)函數(shù)，用于處理異常情況。和函數(shù)執(zhí)行類似，出現(xiàn)錯(cuò)誤之后，Observable 就不再繼續(xù)返回?cái)?shù)據(jù)了。

subscribe() 方法還支持另一種形式傳入回調(diào)函數(shù)：

foo.subscribe({
  next(value) { console.log(value) },
  error(e) { console.log("error: " + e) }
})

而這種形式下，傳入的對(duì)象和 Observable 內(nèi)部執(zhí)行函數(shù)中的 observer 參數(shù)在形式上就比較一致了。

Observable 內(nèi)部的邏輯可以異步多個(gè)返回值，甚至返回?zé)o數(shù)個(gè)值：

var foo = Rx.Observable.create((observer) => {
  let i = 0
  setInterval(() => observer.next(i++), 1000)
})

foo.subscribe(i => console.log(i))
// 輸出：
// 0
// 1
// 2
// ...

上面例子中，Observable 對(duì)象每隔 1 秒會(huì)返回一個(gè)值給調(diào)用方。即使調(diào)用方不再需要數(shù)據(jù)，仍舊會(huì)繼續(xù)通過(guò)回調(diào)函數(shù)向調(diào)用推送數(shù)據(jù)。

RxJS 提供了中止 Observable 執(zhí)行的機(jī)制：

var foo = Rx.Observable.create((observer) => {
  console.log("start")
  let i = 0
  let timer = setInterval(() => observer.next(i++), 1000)
  return () => {
    clearInterval(timer)
    console.log("end")
  }
})

var subscription = foo.subscribe(i => console.log(i))
setTimeout(() => subscription.unsubscribe(), 2500)
// 輸出：
// start
// 0
// 1
// 2
// end

subscribe() 方法返回一個(gè)訂閱對(duì)象（subscription），該對(duì)象上的 unsubscribe() 方法用于取消訂閱，也就是中止 Observable 內(nèi)部邏輯的執(zhí)行，停止返回新的數(shù)據(jù)。

對(duì)于具體的 Observable 對(duì)象是如何中止執(zhí)行，則要由 Observable 在執(zhí)行后返回一個(gè)用于中止執(zhí)行的函數(shù)，像上面例子中的這種方式。

Observable 執(zhí)行結(jié)束后，會(huì)觸發(fā)觀察者的 complete 回調(diào)，所以可以這樣：

foo.subscribe({
  next(value) { console.log(value) },
  complete() { console.log("completed") }
})

Observable 的觀察者共有上面三種回調(diào)：

next：獲得數(shù)據(jù)

error：處理異常

complete：執(zhí)行結(jié)束

其中 next 可以被多次調(diào)用，error 和 complete 最多只有一個(gè)被調(diào)用一次（任意一個(gè)被調(diào)用后不再觸發(fā)其他回調(diào)）。

對(duì)于函數(shù)返回值，有時(shí)候我們要轉(zhuǎn)換后再使用，例如：

function foo() {
  return 1
}

console.log(f00() * 2)
// 輸出：
// 2

對(duì)于 Observable 返回的值，也會(huì)有類似的情況，不過(guò)通常采用下面的方式：

var foo = Rx.Observable.create((observer) => {
  let i = 0
  setInterval(() => observer.next(i++), 1000)
})

foo.map(i => i * 2).subscribe(i => console.log(i))
// 輸出：
// 0
// 2
// 4
// ...

其實(shí) foo.map() 返回了新的 Observable 對(duì)象，上面代碼等價(jià)于：

var foo2 = foo.map(i => i * 2)
foo2.subscribe(i => console.log(i))

Observable 對(duì)象 foo2 被訂閱時(shí)執(zhí)行的內(nèi)部邏輯可以簡(jiǎn)單視為：

function subscribe(observer) {
  let mapFn = v => v * 2
  foo.subscribe(v => {
    observer.next(mapFn(v))
  })
}

將這種對(duì)數(shù)據(jù)的處理和數(shù)組進(jìn)行比較看看：

var array = [0, 1, 2, 3, 4, 5]
array.map(i => i * 2).forEach(i => console.log(i))

是不是有點(diǎn)像？

除了 map() 方法，Observable 還提供了多種轉(zhuǎn)換方法，如 filter() 用于過(guò)濾數(shù)據(jù)，find() 值返回第一個(gè)滿足條件的數(shù)據(jù)，reduce() 對(duì)數(shù)據(jù)進(jìn)行累積處理，在執(zhí)行結(jié)束后返回最終的數(shù)據(jù)。這些方法和數(shù)組方法功能是類似的，只不過(guò)是對(duì)異步返回的數(shù)據(jù)進(jìn)行處理。還有一些轉(zhuǎn)換方法更加強(qiáng)大，例如可以 debounceTime() 可以在時(shí)間維度上對(duì)數(shù)據(jù)進(jìn)行攔截等等。

Observable 的轉(zhuǎn)換方法，本質(zhì)不過(guò)是創(chuàng)建了一個(gè)新的 Observable，新的 Observable 基于一定的邏輯對(duì)原 Observable 的返回值進(jìn)行轉(zhuǎn)換處理，然后再推送給觀察者。

Observable 就是一個(gè)奇怪的函數(shù)，它有和函數(shù)類似的東西，例如封裝了一段邏輯，每次調(diào)用時(shí)都會(huì)重新執(zhí)行邏輯，執(zhí)行有返回?cái)?shù)據(jù)等；也有更特殊的特性，例如數(shù)據(jù)是推送（push）的方式返回給調(diào)用方法，返回值可以是異步，可以返回多個(gè)值等。

不過(guò)將 Observable 視作特殊函數(shù)，至少對(duì)于理解 Observable 上是比較有幫助的。

Observable 也被視為 data stream（數(shù)據(jù)流），這是從 Observable 可以返回多個(gè)值的角度來(lái)看的，而數(shù)據(jù)轉(zhuǎn)換則是基于當(dāng)前數(shù)據(jù)流創(chuàng)建新的數(shù)據(jù)流，例如：

不過(guò)上圖看到的只是數(shù)據(jù)，而將 Observable 視為特殊函數(shù)時(shí)，不應(yīng)該忘了其內(nèi)部邏輯，不然數(shù)據(jù)是怎么產(chǎn)生的呢。