淺談前端響應式設計（二）

Observable是一個集合了觀察者模式、迭代器模式和函數式的庫，提供了基於事件流的強大的異步處理能力，而且已在 Stage1草案中。介紹的 Rxjs是 Observable的一個實現，它是ReactiveX衆多語言中的 JavaScript版本。

在 JavaScript中，咱們可使用 T|null去處理一個單值，使用 Iterator去處理多個值得狀況，使用 Promise處理異步的單個值，而 Observable則填補了缺失的「異步多個值」。

使用 Rxjs

上文提到使用 EventEmitter作響應式處理，在 Rxjs中稍有些不一樣：

/*

const change$ = new Subject();

<Input change$={change$} />

<Search change$={change$} />

*/

class Input extends Component {
  state = {
    value: ''
  };

  onChange = e => {
    this.props.change$.next(e.target.value);
  };

  componentDidMount() {
    this.subscription = this.props.change$.subscribe(value => {
      this.setState({
        value
      });
    });
  }

  componentWillUnmount() {
    this.subscription.ubsubscribe();
  }

  render() {
    const { value } = this.state;

    return <input value={value} onChange={this.onChange} />;
  }
}

class Search extends Component {
  // ...

  componentDidMount() {
    this.subscription = this.props.change$.subscribe(value => {
   ajax(/* ... */).then(list =>
        this.setState({
          list
        })
      );
    });
  }

  componentWillUnmount() {
    this.subscription.ubsubscribe();
  }

  render() {
    const { list } = this.state;

    return <ul>{list.map(item => <li key={item.id}>{item.value}</li>)}</ul>;
  }
}
複製代碼

在這裏，咱們雖然也須要手動釋放對事件的訂閱，可是得益於 Rxjs的設計，咱們不須要像 EventEmitter那樣去存下回調函數的實例，用於釋放訂閱，所以咱們很容易就能夠經過高階組件解決這個問題。例如：

const withObservables = observables => ChildComponent => {
  return class extends Component {
    constructor(props) {
      super(props);
      this.subscriptions = {};
      this.state = {};
      Object.keys(observables).forEach(key => {
        this.subscriptions[key] = observables[key].subscribe(value => {
          this.setState({
            [key]: value
          });
        });
      });
    }

    onNext = (key, value) => {
      observables[key].next(value);
    };

    componentWillUnmount() {
      Object.keys(this.subscriptions).forEach(key => {
        this.subscriptions[key].unsubscribe();
      });
    }

    render() {
      return (
        <ChildComponent {...this.props} {...this.state} onNext={this.onNext} />
      );
    }
  };
};
複製代碼

這樣在須要聚合多個數據源時，也不會像 EventEmitter那樣手動釋放資源形成麻煩。同時，在 Rxjs中咱們還有專用於聚合數據源的方法：

Observable.combineLatest(foo$, bar$)
  .pipe(
      // ...
  );
複製代碼

顯然相對於 EventEmitter的方式十分高效，同時它相對於 Mobx也有巨大的優點。在 Mobx中，咱們提到須要聚合多個數據源的時候，採用 autoRun的方式容易收集到沒必要要的依賴，使用 observe則不夠高效。在 Rxjs中，顯然不會有這些問題， combineLatest能夠以很簡練的方式聲明須要聚合的數據源，同時，得益於 Rxjs設計，咱們不須要像 Mobx一個一個去調用 observe返回的析構，只須要處理每個 subscribe返回的 subscription：

class Foo extends Component {
  constructor(props) {
    super(props);
    this.subscription = Observable.combineLatest(foo$, bar$)
      .pipe(
        // ...
      )
      .subscribe(() => {
        // ...
      });
  }

  componentWillUnmount() {
    this.subscription.unsubscribe();
  }
}
複製代碼

異步處理

Rxjs使用操做符去描述各類行爲，每個操做符會返回一個新的 Observable，咱們能夠對它進行後續的操做。例如，使用 map操做符就能夠實現對數據的轉換：

foo$.map(event => event.target.value);
複製代碼

Rxjs5.5以後全部的 Observable上都引入了一個 pipe方法，接收若干個操做符， pipe方法會返回一個 Observable。所以，咱們能夠很容易配合 tree shaking實現對操做符的按需引入，而不是把整個 Rxjs引入進來：

import { map } from 'rxjs/operators';

foo$.pipe(map(event => event.target.value));
複製代碼

推薦使用這種寫法。 在討論面向對象的響應式的響應式中，咱們提到對於異步的問題，面向對象的方式很差處理。在 Observable中咱們能夠經過 switchMap操做符處理異步問題，一個異步搜索看起來會是這樣：

input$.pipe(switchMap(keyword => Observable.ajax(/* ... */)));
複製代碼

在處理異步單值時，咱們可使用 Promise，而 Observable用於處理異步多個值，咱們能夠很容易把一個 Promise轉成一個 Observable，從而複用已有的異步代碼：

input$.pipe(switchMap(keyword => fromPromise(search(/* ... */))));
複製代碼

switchMap接受一個返回 Observable的函數做爲參數，下游的流就會切到這個返回的 Observable。 而要聚合多個數據源並作異步處理時：

combineLatest(foo).pipe( switchMap(keyword => fromPromise(someAsyncOperation(/* ... */))) ); 同時，因爲標準制定的 Promise是沒有 cancel方法的，有時候咱們要取消異步方法的時候就有些麻煩（主要是爲了解決一些併發安全問題）。 switchMap當上遊有新值到來時，會忽略結束已有未完成的 Observable而後調用函數返回一個新的 Observable，咱們只使用一個函數就解決了併發安全問題。固然，咱們能夠根據實際須要選用 switchMap、 mergeMap、 concatMap、 exhaustMap等。

而對於時間軸的操做， Rxjs也有巨大優點。上篇博客中提到當咱們須要延時 5 秒作操做時，不管是 EventEmitter仍是面向對象的方式都力不從心，而在 Rxjs中咱們只須要一個 delay操做符便可解決問題：

input$.pipe(
  delay(5000) // 下游會在input$值到來後5秒才接到數據
);
複製代碼

用 Rxjs 處理數據

在實際開發過程當中，事件不能解決全部問題，咱們每每會須要存儲數據，而 Observable被設計成用於處理事件，所以它有不少符合事件直覺的設計。

Observable被設計爲懶（ lazy）的，噹噹沒有訂閱者時，一個流不會執行。對於事件而言，沒有事件的消費者那麼不執行也不會有問題。而在 GUI 中，訂閱者多是 View：

class View extends Component {
  state = {
    input: ''
  };

  componentDidMount() {
    this.subscription = input$.subscribe(input => {
      this.setState({
        input
      });
    });
  }

  componentWillUnmount() {
    this.subscription.unsubscribe();
  }

  render() {
    // ...
  }
}
複製代碼

因爲這個 View可能不存在，例如路由被切走了，那麼咱們的事件源就沒有了訂閱者，他就不會運行。可是咱們但願在路由被且走後，後臺的數據依然會繼續。

對於事件而言，在事件發生以後的訂閱者不會受到訂閱以前的邏輯。例如在 EventEmitter中：

eventEmitter.emit('hello', 1);
// ...
eventEmitter.on('hello', function listener() {});
複製代碼

因爲 listener是在 hello事件發生後在監聽的，不會收到值爲 1的事件。可是這在處理數據的時候會形成麻煩，咱們的數據在 View被卸載（例如路由切走）後丟失。

同時，因爲 Observable沒有提供直接取到內部狀態的方法，當咱們使用 Observable處理數據時，咱們不方便隨時拿到數據。那有辦法解決這個問題，從而使 Observable強大抽象能力去賦能數據層呢？

回到 Redux。 Redux的事件（Action）實際上是一個事件流，那麼咱們就能夠很天然地把 Redux的事件流融入到 Rxjs流中：

() => next => {
  const action$ = new Subject();

  return action => {
    action$.next(action);
    // ...
  };
};
複製代碼

經過這樣的封裝，redux-observable就能讓咱們把 Observable強大的事件描述和處理能力和 Redux結合。咱們能夠很是方便地根據 Action去處理反作用：

action$.pipe(
  ofType('ACTION_1'),
  switchMap(() => {
    // ...
  }),
  map(res => ({
    type: 'ACTION_2',
    payload: res
  }))
);

action$.pipe(
  ofType('ACTION_3'),
  mergeMap(() => {
    // ...
  }),
  map(res => ({
    type: 'ACTION_4',
    payload: res
  }))
);
複製代碼

ReduxObservable使咱們能夠結合 Redux和 Observable。在這裏， Action被視做一個流， ofType至關於 filter(action=>action.type==='SOME_ACTION')，從而獲得須要監聽的 Action，得益於 Redux的設計，咱們能夠經過監聽 Action去完成反作用的處理或者監聽數據變化。最後這個流返回一個新的 Action流， ReduxObservable會把這個新的 Action流中的 Action dispatch出去。由此，咱們在使用 Redux存儲數據的基礎上得到了 Rxjs對異步事件的強大處理能力。