基于Redux-Observable的同步系统：首个待处理任务后跳过Observable发射

我来给你梳理下怎么实现这个需求哈——既要让同步请求按顺序排队执行（类似concatMap的行为），又要处理重复请求的情况。结合redux-observable的特性，咱们可以通过操作符的组合来搞定：

一、基础版：仅实现请求排队 #

如果你的需求只是“同步执行时新请求自动排队，一个完成再执行下一个”，用concatMap就可以直接实现，代码示例如下：

import { ofType } from 'redux-observable';
import { concatMap, map, catchError } from 'rxjs/operators';
import { of } from 'rxjs';
import { performSync } from './your-sync-utils'; // 替换成你实际执行同步的Observable

// 对应的Epic
const syncEpic = (action$) => action$.pipe(
  // 监听你的同步请求动作类型
  ofType('SYNC_REQUEST'),
  // concatMap会把每个请求按顺序放入队列，前一个同步完成后再执行下一个
  concatMap((action) => 
    performSync(action.payload).pipe(
      // 同步成功后分发成功动作
      map(() => ({ type: 'SYNC_SUCCESS', payload: action.payload })),
      // 处理同步失败的情况
      catchError((error) => of({ 
        type: 'SYNC_FAILURE', 
        payload: error, 
        meta: { originalRequest: action.payload } 
      }))
    )
  )
);

这个版本里，所有SYNC_REQUEST动作都会被按顺序处理，哪怕同步正在执行，新的请求也会乖乖排队等着。

二、进阶版：排队+过滤重复请求 #

如果你的场景中会出现多个完全相同的同步请求，不想让它们都占着队列位置，就可以加上distinctUntilChanged来过滤连续重复的请求：

import { ofType } from 'redux-observable';
import { concatMap, distinctUntilChanged, map, catchError } from 'rxjs/operators';
import { of } from 'rxjs';
import { performSync } from './your-sync-utils';

const syncEpic = (action$) => action$.pipe(
  ofType('SYNC_REQUEST'),
  // 这里根据请求的payload判断是否重复，你可以根据实际需求调整判断逻辑
  distinctUntilChanged((prevAction, currAction) => 
    JSON.stringify(prevAction.payload) === JSON.stringify(currAction.payload)
  ),
  concatMap((action) => 
    performSync(action.payload).pipe(
      map(() => ({ type: 'SYNC_SUCCESS', payload: action.payload })),
      catchError((error) => of({ 
        type: 'SYNC_FAILURE', 
        payload: error, 
        meta: { originalRequest: action.payload } 
      }))
    )
  )
);

这个版本里，连续的相同请求只会被保留第一个，后面的重复请求会被直接过滤，不会进入排队队列。

三、高阶版：排队+保留最新重复请求 #

如果你的需求是“当队列里已经有某个请求在等待时，新的相同请求替换掉旧的，只执行最新的那个”，可以用concatMap结合switchMap和takeUntil来实现：

import { ofType } from 'redux-observable';
import { concatMap, switchMap, takeUntil, distinctUntilChanged, map, catchError } from 'rxjs/operators';
import { of, last } from 'rxjs';
import { performSync } from './your-sync-utils';

const syncEpic = (action$) => action$.pipe(
  ofType('SYNC_REQUEST'),
  // 按请求分组，只处理每个分组内的最新请求
  concatMap((initialAction) => 
    action$.pipe(
      ofType('SYNC_REQUEST'),
      distinctUntilChanged((a, b) => JSON.stringify(a.payload) === JSON.stringify(b.payload)),
      // 直到当前初始请求对应的同步开始执行，停止监听新的重复请求
      takeUntil(performSync(initialAction.payload).pipe(map(() => true))),
      // 取这个时间段内的最后一个请求（最新的那个）
      last(),
      // 执行最新的同步请求
      switchMap((latestAction) => 
        performSync(latestAction.payload).pipe(
          map(() => ({ type: 'SYNC_SUCCESS', payload: latestAction.payload })),
          catchError((error) => of({ 
            type: 'SYNC_FAILURE', 
            payload: error, 
            meta: { originalRequest: latestAction.payload } 
          }))
        )
      )
    )
  )
);

这个逻辑的核心是：每个正在排队的请求组里，只保留最新的那个请求去执行，避免无效的重复同步操作。

你可以根据自己的实际业务场景，选择最适合的方案~

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