React实现队列解决多个请求频繁并发到达server时序乱序问题

背景：前端存在多个请求并发到后端，默认览器运行最多6并发请求，导致很多请求padding状态、而且后端处理不过来；

现象：

场景1：如果这些数据可用丢包，使用节流就可用解决；- 简单不做赘述；

场景2：如果这些数据不能丢包，需要把数据合并再发送给服务器。

要求：加入需要等待服务器处理完再发下一包数据，支持多个请求任务队列；

针对场景2问题需要解决以下三个：

1、请求（任务）上一个请求完成，再开始执行下个任务；

2、队列中任务参数合并再发送请求；

3、支持多个任务在队列中进工作；

定义任务类型：

/** * 任务状态 */ export enum TaskStateE { unStart = 0, // 未开始 start, // 执行中 end, // 执行完成 } /** * 任务对象 * @template T 任务参数类型 */ export interface TaskItemT<T = unknown> { taskName: string; // 任务名称 id: string | number; // 任务id state: TaskStateE; // 任务状态 count: number; // 执行次数 param: T; // 任务参数 startTime?: number; // 任务开始时间 endTime?: number; // 任务完成时间 start: (task: TaskItemT<T>) => Promise<unknown>; // 开始执行 /** * 可选的合并策略，如果提供则使用该策略合并相同 taskName 的任务 */ mergeStrategy?: ITaskMergeStrategy<T>; }

整体方案设计：

任务添加流程：

自定义Hook: useQueueTask.tsx

import { useMemoizedFn, useUnmount } from 'ahooks'; import { useRef } from 'react'; import type { TaskItemT } from '@/enum/common'; import { TaskStateE, TIMER_TIME } from '@/enum/common'; /** * 通用任务队列 Hook * * @description 提供任务队列管理功能，支持任务添加、合并、执行和更新 * @param options 配置选项 * @returns 任务队列操作方法 * * @example * ```tsx * const { addTask, updateTask, clearQueue } = useQueueTask({ enableDebugLog: true }); * * addTask({ * taskName: 'fetchData', * id: '1', * state: TaskStateE.unStart, * count: 0, * param: { userId: 1 }, * start: (task) => { * // 执行任务逻辑 * }, * }); * ``` */ function useQueueTask() { // 任务队列 const queue = useRef<TaskItemT[]>([]); // 任务定时器 const taskScheduler = useRef<ReturnType<typeof setInterval> | null>(null); // 正在执行的任务 const executingTask = useRef<TaskItemT | undefined>(undefined); /** * 停止任务调度器 * @description 清除定时器 */ const stopScheduler = useMemoizedFn(() => { if (taskScheduler.current !== null) { clearInterval(taskScheduler.current); taskScheduler.current = null; } }); /** * 合并任务 * @description 根据合并策略合并已存在的任务和新任务 * @param existingTask 已存在的任务 * @param newTask 新任务 * @returns 合并后的任务 */ const mergeTask = <T = unknown,>(existingTask: TaskItemT<T>, newTask: TaskItemT<T>): TaskItemT<T> => { if (newTask.mergeStrategy) { // 使用提供的合并策略 const mergedParam = newTask.mergeStrategy.merge(existingTask.param as T, newTask.param); return { ...existingTask, param: mergedParam, start: newTask.start, // 使用新任务的 start 方法 mergeStrategy: newTask.mergeStrategy, }; } // 没有合并策略，直接返回新任务（替换） return newTask; }; /** * 添加任务 * @description 将任务添加到队列中，如果存在相同 taskName 的任务，则根据合并策略处理 * @param task 要添加的任务 */ const addTask = useMemoizedFn(<T = unknown,>(task: TaskItemT<T>) => { const index = queue.current.findIndex(item => item.taskName === task.taskName); if (index !== -1) { // 如果存在相同 taskName 的任务，使用合并策略 const existingTask = queue.current[index] as TaskItemT<T>; const mergedTask = mergeTask(existingTask, task); queue.current.splice(index, 1, mergedTask); } else { // 不存在相同 taskName 的任务，直接添加 queue.current.push(task as TaskItemT); } // 如果任务队列有任务，则开始执行任务 startScheduler(); }); /** * 执行任务 * @description 执行任务并处理错误 * @param task 要执行的任务 */ const executeTask = (task: TaskItemT) => { // 原子性操作：先设置任务和状态，避免竞态条件 // 必须在调用 task.start 之前设置，确保后续的 processTask 调用能够检测到正在执行的任务 if (executingTask.current !== undefined && executingTask.current.id !== task.id) { // 如果当前已经有其他任务在执行，不应该执行此任务（这种情况理论上不应该发生，但作为防御性编程） console.warn('Task execution skipped: another task is already executing', { currentTask: executingTask.current.id, newTask: task.id, }); // 将任务重新放回队列头部 queue.current.unshift(task); return; } // 设置执行中的任务标记（原子操作） executingTask.current = task; task.state = TaskStateE.start; task.startTime = Date.now(); // 确保 Promise 处理：将返回值转换为 Promise // 使用 Promise.resolve 确保即使 task.start 返回同步值也能正确处理 let taskPromise: Promise<any>; try { const result = task.start(task); taskPromise = Promise.resolve(result); } catch (error) { // 如果 task.start 同步抛出错误，立即处理 taskPromise = Promise.reject(error); } // 在 finally 中统一处理任务完成逻辑 taskPromise .finally(() => { // 只有当前任务还在执行时才更新状态 // 这个检查很重要，防止任务完成时被其他任务覆盖（虽然理论上不应该发生） if (executingTask.current?.id === task.id) { task.state = TaskStateE.end; task.endTime = Date.now(); // 清空正在执行的任务，确保下一个任务可以执行 // 必须在所有状态更新完成后才能清空，确保原子性 executingTask.current = undefined; } }) .catch(error => { // 错误已经在 finally 中处理，这里只记录日志 // 不再重新抛出，因为 finally 已经处理了状态清理 console.error('Task execution failed:', error); }); }; /** * 任务调度 * @description 从队列中取出任务并执行，支持批量处理相同名称的任务（取最后一个） */ const processTask = useMemoizedFn(() => { // 严格检查：只有在没有正在执行的任务时才能处理新任务 // executingTask.current === undefined 表示当前没有任务在执行 if (executingTask.current !== undefined) { return; } // 检查队列是否有任务 if (queue.current.length === 0) { return; } // 获取队首任务的名称 const firstTaskName = queue.current[0]?.taskName; if (!firstTaskName) { return; } // 取队首任务 const task = queue.current.shift(); if (task) { // 双重检查锁模式（Double-Checked Locking） // 在取出任务后、执行前再次检查，确保原子性 // 这是为了防止多个 processTask 调用同时通过第一次检查 if (executingTask.current !== undefined) { // 如果此时已经有任务在执行，将任务重新放回队列头部 queue.current.unshift(task); return; } // 执行任务（executeTask 内部会立即设置 executingTask.current，确保原子性） executeTask(task); } }); /** * 清空队列 * @description 停止调度器、清空队列和正在执行的任务 */ const clearQueue = useMemoizedFn(() => { stopScheduler(); queue.current.length = 0; executingTask.current = undefined; }); /** * 启动任务调度器 * @description 如果调度器未启动，则启动定时器 */ const startScheduler = useMemoizedFn(() => { // 如果定时器已经启动，不需要重复启动 if (taskScheduler.current !== null) { return; } taskScheduler.current = setInterval(() => { // 始终尝试处理任务（processTask 内部会检查是否可以执行） processTask(); // 如果队列为空且没有正在执行的任务，停止定时器 if (queue.current.length === 0 && executingTask.current === undefined) { stopScheduler(); } }, TIMER_TIME); }); // 组件卸载时清理资源 useUnmount(() => { clearQueue(); }); return { addTask }; } export default useQueueTask;

业务代码调用添加任务：

addTask({ id: uuidv4(), taskName: 'ledScreenPainterClearApi', state: TaskStateE.unStart, count: 0, param: {a: 1}, start: (task: TaskItemT<{ a: number }>) => { return ledScreenPainterClearApi(task.param); }, });