RxJS+Redux-Observable：如何让多个Action串行执行而非并行

问题核心 #

当前发牌逻辑会并行触发所有玩家的抽牌操作，导致所有抽牌读取的都是初始牌组状态，前一位玩家抽牌后的牌组更新无法被后续玩家感知。

现有代码分析 #

发牌Epic（问题所在） #

export const dealCardsEpic = (action$: any, state$: any) => action$.pipe(
  filter(dealCards.match),
  mergeMap(() => {
    const players = selectPlayers(state$.value);

    return of(...players.inGame.flatMap((player, index) => {
      return drawCards({ playerId: player.id, count: index === 0 ? 5 : 4})
    }));
  })
);

这里使用of(...players)会一次性将所有drawCards action同步派发，相当于并行执行抽牌，所有操作共享初始的state$.value。

抽牌Epic #

export const drawFromDeckEpic = (action$: any, state$: any) => action$.pipe(
  filter(drawCards.match),
  concatMap((action: any) => {
    const cards = selectTopCards(state$.value, action.payload.count);
    
    return [
      addCardsToPlayer({playerId: action.payload.playerId, cards}),
      removeCards(action.payload.count),
      setNextPlayerTurn()
    ];
  }),
);

抽牌Epic内部用了concatMap，但因为上游的drawCards是并行派发的，所以每个抽牌操作还是读取旧状态。

解决方案 #

1. 修改发牌Epic，实现顺序派发抽牌 #

将mergeMap替换为concatMap，并通过监听前一次抽牌完成的action（比如deck/removeCards）来触发下一次抽牌，确保顺序执行：

export const dealCardsEpic = (action$: any, state$: any) => action$.pipe(
  filter(dealCards.match),
  concatMap(() => {
    const players = selectPlayers(state$.value);
    
    // 从玩家数组创建Observable，逐个顺序处理
    return from(players.inGame).pipe(
      concatMap((player, index) => 
        action$.pipe(
          // 监听前一次抽牌完成的标记action
          filter(action => action.type === 'deck/removeCards'),
          take(1),
          // 第一个玩家无需等待，直接触发
          startWith(null),
          map(() => drawCards({
            playerId: player.id,
            count: index === 0 ? 5 : 4
          }))
        )
      ),
      // 手动触发第一个玩家的抽牌
      startWith(drawCards({
        playerId: players.inGame[0].id,
        count: 5
      }))
    );
  })
);

2. 优化抽牌Epic，确保读取最新状态 #

在抽牌Epic中使用withLatestFrom获取实时状态，避免读取快照式的旧状态：

export const drawFromDeckEpic = (action$: any, state$: any) => action$.pipe(
  filter(drawCards.match),
  concatMap((action) => 
    state$.pipe(
      take(1),
      map(state => selectTopCards(state, action.payload.count)),
      map(cards => [
        addCardsToPlayer({playerId: action.payload.playerId, cards}),
        removeCards(action.payload.count),
        setNextPlayerTurn()
      ]),
      mergeAll()
    )
  )
);

预期效果 #

修改后DevTools中的Action顺序会变为：

• deck/dealCards
• deck/drawCards
• players/addCardsToPlayer
• deck/removeCards
• players/setNextPlayersTurn
• deck/drawCards
• players/addCardsToPlayer
• deck/removeCards

每个玩家的抽牌操作都会等待前一个玩家的抽牌副作用完成，牌组状态会实时更新，确保发牌逻辑符合顺序要求。

内容的提问来源于stack exchange，提问作者Petr Velký