RxJS与Redux结合使用（一）：打造自己的redux-observable

到此，我们的中间件已经有初步处理异步逻辑的能力，但是，在现实中，我们的异步逻辑不可能只有一个，所以epic是会有很多的，而store去订阅的流只能是一个，所以这么多的epic产生的流要合并成一个流。

合并流的操作，强大的RxJs自然是有安排的，相信你想到了操作符merge，我们可以提供一个combineEpics的函数：

export%20const%20combineEpics%20=%20(...epics)%20=>%20{
%20%20const%20merger%20=%20(...args)%20=>%20merge(
%20%20%20%20...epics.map((epic)%20=>%20{
%20%20%20%20%20%20const%20output$%20=%20epic(...args);
%20%20%20%20%20%20return%20output$;
%20%20%20%20})
%20%20);
%20%20return%20merger;
};

上面的代码不难理解，combineEpics整合了所有传入的epic，然后返回一个merger，这个merger是利用merge操作符，将所有的epic产生的流合并成一个流。

代码形式为：

const pingEpic = action$ => action$.pipe(
  filter(action => action.type === 'ping'),
  map(() => ({ type: 'pong' })),
);

const getUserEpic = action$ => action$.pipe(
  filter(action => action.type === 'GET_USER'),
  map(() => ({ type: 'GET_USER_SUCCESS', payload: { user: { name: 'kang' } } })),
);

const rootEpic = combineEpics(pingEpic, getUserEpic);

export default (store) => {
  const action$ = new Subject();
  const newAction$ = rootEpic(action$);

  newAction$.subscribe(store.dispatch);
  return next => (action) => {
    const result = next(action);
    action$.next(action);
    return result;
  };
};

state获取

在epic中我们不可避免地要借助state里面的数据进行不同的处理，所以我们是需要获取到state的，所以你可以在中间件中的epci执行函数中添加一个参数，将state获取函数暴露出去：

export default (store) => {
  ...
  const newAction$ = rootEpic(action$, store.getState);
  ...
};

这样epic里就可以用getState()获取state：

const pingEpic = (action$, getState) => action$.pipe(
  filter(action => action.type === 'ping'),
  map(() => ({ type: 'pong', payload: getState() })),
);

进一步优化：将state也流化

上面的做法是直接去获取state，这样的做法是主动获取，不符合函数响应式编程模式。函数响应式中，state的改变状态，应该是要能被观察的。

当state也能被响应观察，我们就可以做更多的功能，例如：当state的某些数据在发生变化的时候，我们要去进行实时保存。

在传统模式的做法中，你可以在中间件中这样写：

export default store => next => action => {
    const oldState = store.getState();
    const result = next(action);
    const newState = store.getState();
    // 类似这样的写法
    if (newState.xxx !== oldState.xxx) {
        fetch('/api/save', {
            method: 'POST',
            body: {
            
            }
        }).then(() => {}).catch(() => {})
    }
    return result;
}

这个处理逻辑要独立为一个中间件，而如果你将state也流化，你可以直接使用epic这样处理：

const saveEpic = (action$, state$) => state$.pipe(
const autoSaveEpic = (action$, state$) =>
  return action$.pipe(
    filter(action => action.type === 'AUTO_SAVE_ENABLE'), // 自动保存的开启
    exhaustMap(() => state$.pipe(
        pluck('xxx'), // 获取state.xxx
        distinctUntilChanged(), // 前后值不同时才将其发出。
        concatMap((value) => {
            // fetch to save
        }),
        // 自动保存的关闭
        takeUntil(action$.pipe(
            filter(action => action.type === 'AUTO_SAVE_DISABLE')
        ))
    ))
  )
)

如果仔细阅读这段代码，可以发现这样的方式可以使用非常优雅的方式控制这个自动保存，可以和action$结合使用，快速开关自动保存，可以利用RxJs的特性解决保存的异步执行延迟问题。

如果你只是单存想要获取最新state，可以使用withLatestFrom操作符：

const countEpic = (action$, state$) => action$.pipe(
  filter(action => action.type === 'count'),
  withLatestFrom(state$),
  switchMap(([action, state]) => {
    return of({ type: 'whatever' });
  })
);

在中间件加入state流：

export default (store) => {
  const action$ = new Subject();
  const state$ = new Subject();
  const source$ = rootEpic(action$, state$);

  source$.subscribe(store.dispatch);
  return next => (action) => {
    const result = next(action);
    state$.next(store.getState());
    action$.next(action);
    return result;
  };
};

注意state最新值不是默认state。

Action的顺序问题

如果你有耐心看到这里，那么说明你对于redux结合RxJs的使用已经理解得差不多了，但是这里还是有个问题，就是action的生效顺序，我们可以直接看个例子说明，假设有下面这样两个epic：

const epic1 = action$ => action$.pipe(
  filter(action => action.type === 'one'),
  mergeMap(() => of({ type: 'two' }, { type: 'three' })),
);

const epic2 = action$ => action$.pipe(
  filter(action => action.type === 'two'),
  mergeMap(() => of({ type: 'four' })),
);

当 store.dispatch({ type: 'one' }) 的时候，action的顺序为：

'one' -> 'two' -> 'four' -> 'three'

可见，action的执行顺序并不是如我们预期的那样，在two触发后就发出了four，这是因为RxJs默认的调度器是同步的，用一段简单的代码，上面的效果类似于：

class Print {
  constructor(name, nexts = []) {
    this.name = name;
    this.nexts = nexts;
  }
  print() {
    console.log(this.name);
    this.nexts.forEach((p) => {
      p.print();
    });
  }
}
const three = new Print('three');
const four = new Print('four');
const two = new Print('two', [four]);
const one = new Print('one', [two, three]);
one.print(); // one, two, four, three

换成上面的代码的话你就不陌生了吧，也会对于输出的结果表示肯定，但是我们需要的效果是

'one' -> 'two' -> 'three' -> 'four'

这如何做到？

明显，需要将调度器换成其他的，RxJs有这么几种调度器：null(同步)、asap、queue、async、animationFrame。最后一种是动画场景的调度器，直接剔除，默认是第一种，那么就剩下asap、queue、async。在这个场景下，这三种调度器都是可行的，但是queue在大量的数据的时候对于性能是有利的，所以这里可以使用它。不过，记住，这三种调度器是有区别的，大家有兴趣的自己去google一下，只提示：asap是Micro Task、async是Macro Task、queue在延迟为0的时候接近于同步，在延迟不为0的时候与async一样。

中间件中：

  const action$ = new Subject().pipe(
    observeOn(queue)
  );

这样得到的结果就是：

'one' -> 'two' -> 'three' -> 'four'

如果用简单的代码，相当于发生了这样的变化：

class Print {
  constructor(name, nexts = []) {
    this.name = name;
    this.nexts = nexts;
  }
  print() {
    console.log(this.name);
    this.nexts.forEach((p) => {
      setTimeout(() => p.print(), 0);
    });
  }
}
const three = new Print('three');
const four = new Print('four');
const two = new Print('two', [four]);
const one = new Print('one', [two, three]);
one.print(); // one, two, three, four

总结

本文就讲到这里，这次介绍了如何自己实现一个redux-observable，下次会讲redux-observable在实战中的一些应用，例如怎么类似dva那样进行模块化开发、如何统一处理loading、error等。