RxJS升级V6后Redux-Observable主Epic动作顺序异常问题

我使用React、Redux、Redux-Observable和RxJS技术栈，主Epic中包含多个动作及一个内部Epic，但内部Epic的action2未按预期在action1和action3之间执行，而是在主Epic末尾执行。该问题出现在将RxJS从V5升级到V6之后。

代码实现 #

主Epic代码 #

import { Observable, from, concat, of, merge, forkJoin } from 'rxjs';
import { ofType } from 'redux-observable';
import { mergeMap, concatMap, switchMap, map, delay, debounceTime, takeUntil, catchError } from 'rxjs/operators';
import * as actions from '../actions';
import * as types from '../actions/types';

const mainEpic = (action$, store, deps) => {
  return action$.pipe(
    ofType(types.MAIN_EPIC),
    mergeMap( action => of(
      actions.action1(action.payload.oldValue),
      actions.innerEpic(action.payload.newValue),
      actions.action3(action.payload.newValue)
    ))
  )
}
export default mainEpic;

内部Epic代码 #

import { Observable, from, concat, of, merge, forkJoin } from 'rxjs';
import { ofType } from 'redux-observable';
import { mergeMap, concatMap, switchMap, map } from 'rxjs/operators';
import * as actions from '../actions';
import * as types from '../actions/types';

const innerEpic = (action$, store, deps) => {
  return action$.pipe(
    ofType(types.INNER_EPIC),
    switchMap( action =>
      of(
        actions.epicAction2(action.payload)
      )
    )
  )
}

export default innerEpic;

Reducer代码 #

import * as types from "../actions/types";

const defaultState = { data: [] };

export default function(state = defaultState, action) {
  console.log(`    *** reducer controllersTaken called with action type '${action.type}'`)
  try{
    switch (action.type) {
      case types.ACTION_1:
        return state;

      case types.ACTION_2:
        return state;

      case types.ACTION_3:
        return state;

      default: 
        return state;
    }
  } catch (err) {
    console.log('ERROR', err);
    return state;
  }
}

Action定义代码 #

import * as types from '../actions/types';

export const mainEpicCall = payload => {
  console.log(' --- dispatch main-epic');
  return (
    function(dispatch, getState) {
      dispatch({
        type: types.MAIN_EPIC,
        payload
      });
    }
  );
}

export const action1 = payload => {
  console.log(' --- dispatch action1');
  return (
    function(dispatch, getState) {
      dispatch({
        type: types.ACTION_1
      });
    }
  );
}

export const innerEpic = payload => {
  console.log(' --- dispatch inner-epic');
  return (
    function(dispatch, getState) {
      dispatch({
        type: types.INNER_EPIC,
        payload
      });
    }
  );
}

export const action3 = payload => {
  console.log(' --- dispatch action3');
  return (
    function(dispatch, getState) {
      dispatch({
        type: types.ACTION_3,
        payload
      });
    }
  );
}

export const epicAction2 = payload => {
  console.log(' --- dispatch action2');
  return (
    function(dispatch, getState) {
      dispatch({
        type: types.ACTION_2,
        payload
      });
    }
  );
}

动作类型定义 #

export const MAIN_EPIC = 'main_epic';
export const INNER_EPIC = 'inner_epic';

export const ACTION_1 = 'action_1';
export const ACTION_2 = 'action_2';
export const ACTION_3 = 'action_3';

依赖版本 #

"dependencies": {
    "bootstrap": "^5.1.3",
    "normalizr": "^3.6.2",
    "react": "^17.0.2",
    "react-dom": "^17.0.2",
    "react-scripts": "^5.0.0",
    "redux": "^4.1.2",
    "redux-observable": "^2.0.0",
    "redux-thunk": "^2.4.1",
    "rxjs": "^6.6.7"
  },

预期与实际输出 #

预期控制台输出 #

--- dispatch main-epic
   *** reducer controllersTaken called with action type 'main_epic'
--- dispatch action1
--- dispatch inner-epic
   *** reducer controllersTaken called with action type 'action_1'
   *** reducer controllersTaken called with action type 'inner_epic'
--- dispatch action2
   *** reducer controllersTaken called with action type 'action_2'
--- dispatch action3
   *** reducer controllersTaken called with action type 'action_3'

实际控制台输出 #

--- dispatch main-epic
   *** reducer controllersTaken called with action type 'main_epic'
--- dispatch action1
--- dispatch inner-epic
--- dispatch action3
   *** reducer controllersTaken called with action type 'action_1'
   *** reducer controllersTaken called with action type 'inner_epic'
   *** reducer controllersTaken called with action type 'action_3'
--- dispatch action2
   *** reducer controllersTaken called with action type 'action_2'

问题原因与解决方案 #

原因分析 #

主Epic中通过of()同步发出三个thunk action，redux-thunk会依次执行这些thunk：

1. 执行action1的thunk，dispatch ACTION_1
2. 执行innerEpic的thunk，dispatch INNER_EPIC
3. 立刻执行action3的thunk，dispatch ACTION_3

而内部Epic处理INNER_EPIC action并发出action2的过程是异步的（RxJS的流调度存在延迟），导致ACTION_3的reducer处理先于ACTION_2完成。

修改方案 #

在主Epic中使用concat()串联动作流，确保前一个流完成后再执行下一个，核心是等待内部Epic的ACTION_2触发后再发出action3：

修改后的主Epic代码：

import { of, concat } from 'rxjs';
import { ofType, mergeMap, map, take, startWith } from 'rxjs/operators';
import * as actions from '../actions';
import * as types from '../actions/types';

const mainEpic = (action$, store, deps) => {
  return action$.pipe(
    ofType(types.MAIN_EPIC),
    mergeMap(action => 
      concat(
        // 第一步：发出action1
        of(actions.action1(action.payload.oldValue)),
        // 第二步：发出INNER_EPIC并等待ACTION_2完成
        of({ type: types.INNER_EPIC, payload: action.payload.newValue }).pipe(
          mergeMap(innerAction => 
            action$.pipe(
              ofType(types.ACTION_2),
              take(1), // 只取第一个ACTION_2动作
              map(() => null) // 转换为空值，不额外发动作
            )
          ),
          startWith({ type: types.INNER_EPIC, payload: action.payload.newValue }) // 先发出INNER_EPIC触发内部Epic
        ),
        // 第三步：ACTION_2完成后发出action3
        of(actions.action3(action.payload.newValue))
      )
    )
  );
}
export default mainEpic;

同时可以简化Action定义，去掉innerEpic这个thunk，因为主Epic里直接dispatch INNER_EPIC action即可，避免thunk和Epic的混合调度问题。

修改后，执行顺序会严格按照action1 → 内部Epic触发action2 → action3的顺序执行，符合预期。

内容的提问来源于stack exchange，提问作者Dan