异步任务并发控制

通过队列管理异步任务

const taskCount = ref(10)
/** 最大并发数 */
const MAX_CONCURRENT_REQUESTS = 3
/** 当前请求数 */
let currentRequests = 0
/** 等待上传的任务队列 */
const queue: Array<() => Promise<void>> = []

/**
 * 运行任务
 * @param task
 */
const runTask = async (task: () => Promise<void>) => {
    currentRequests++
    try {
        await task()
    } catch (error) {
        console.log('error: ', error)
    } finally {
        currentRequests--
        if (queue.length > 0) {
            const nextTask = queue.shift()
            nextTask && runTask(nextTask)
        }
    }
}

/**
 * 添加任务到队列
 * @param task
 */
const enqueueTask = async (task: () => Promise<void>) => {
    if (currentRequests < MAX_CONCURRENT_REQUESTS) {
        runTask(task)
    } else queue.push(task)
}

/**
 * 发起任务
 */
const handleRequest = () => {
    console.log(`开始 ${taskCount.value} 个任务`)
    for (let i = 0; i < taskCount.value; i++) {
        enqueueTask(async () => {
            await getNumber({ id: i })
            console.log(`任务 ${i} 完成`)
        })
    }
}

第三方库p-limit

使用第三方库p-limit实现

p-limit 用于在 异步任务执行时限制并发数量，防止一次性发出过多请求。

import pLimit from 'p-limit'
const taskCount = ref(10)
/** 最大并发数 */
const MAX_CONCURRENT_REQUESTS = 3
/** 当前请求数 */
const limit = pLimit(MAX_CONCURRENT_REQUESTS)

const handleRequest = () => {
    console.log(`开始 ${taskCount.value} 个任务`)
    for (let i = 0; i < taskCount.value; i++) {
        limit(() => getNumber({ id: i }))
        console.log(`任务 ${i} 完成`)
    }
}

pLimit(MAX_CONCURRENT_UPLOADS) 创建了一个控制器。

limit(() => getNumber({ id: i })) 将上传任务包裹在并发控制器中，确保最大同时上传的任务数不超过设定值。

实现一个pLimit

export const pLimit = (max: number) => {
    const queue: Array<() => Promise<any>> = []
    let currentRequests = 0
    const runTask = async (task: () => Promise<any>) => {
        currentRequests++
        try {
            await task()
        } catch (error) {
            console.log('error: ', error)
        } finally {
            currentRequests--
            if (queue.length > 0) {
                const nextTask = queue.shift()
                nextTask && runTask(nextTask)
            }
        }
    }
    const enqueueTask = async (task: () => Promise<any>) => {
        if (currentRequests < max) {
            runTask(task)
        } else queue.push(task)
    }
    return enqueueTask
}

模拟setInterval轮询

/**
 * 创建一个定时器，每隔ms执行一遍cb，会立即执行cb
 * @param cb - 回调函数
 * @param ms - 每隔多少毫秒执行一次
 * @returns 返回一个取消定时器的函数
 */
export const loopSetInterval = (cb: () => void, ms: number) => {
  let timeoutId: number
  const excute = () => {
    if (timeoutId) {
      clearTimeout(timeoutId)
    }
    try {
      cb()
      timeoutId = setTimeout(excute, ms)
    } catch (e) {
      console.error(e)
    }
  }
  excute()
  return () => clearTimeout(timeoutId)
}

改成不会立即执行

/**
 * 创建一个定时器，每隔ms执行一遍cb，默认不会立即执行cb
 * @param cb - 回调函数
 * @param ms - 每隔多少毫秒执行一次
 * @param {Object} [options] - 可选参数
 * @param {boolean} [options.immediate=false] - 是否立即执行cb
 * @returns {Function} - 一个取消定时器的函数
 */
export const loopSetInterval = (
    cb: () => void,
    ms: number,
    options: { immediate?: boolean } = { immediate: false }
) => {
    let timeoutId: number
    const excute = () => {
        if (timeoutId) {
            clearTimeout(timeoutId)
        }
        try {
            cb()
            timeoutId = setTimeout(excute, ms)
        } catch (e) {
            console.error(e)
        }
    }
    if (options?.immediate) {
        cb()
    } else timeoutId = setTimeout(excute, ms)
    return () => clearTimeout(timeoutId)
}

进一步优化

增加SSR友好、防止在 Node.js、SSR 中出错。

/**
 * 创建一个定时器，每隔ms执行一遍cb，默认不会立即执行cb
 * @param cb - 回调函数
 * @param ms - 每隔多少毫秒执行一次
 * @param {Object} [options] - 可选参数
 * @param {boolean} [options.immediate=false] - 是否立即执行cb
 * @returns {Function} - 一个取消定时器的函数
 */
export const mySetInterval = (
    cb: () => void,
    ms: number,
    options: { immediate?: boolean } = { immediate: false }
): (() => void) => {
    if (typeof window === 'undefined') {
        console.warn('loopSetInterval: 在非浏览器环境中无效')
        return () => {}
    }

    if (ms <= 0) {
        throw new Error('时间间隔必须大于0')
    }

    let timeoutId: number | undefined
    let isRunning = true

    const execute = () => {
        if (!isRunning) return
        try {
            cb()
        } catch (e) {
            console.error('定时器回调执行出错:', e)
        }
        // 只有当isRunning为true时才继续设置下一个定时器
        if (isRunning) {
            timeoutId = window.setTimeout(execute, ms)
        }
    }
    if (options.immediate) {
        try {
            cb()
        } catch (error) {
            console.error('loopSetInterval: 初始回调执行出错', error)
        }
    }

    timeoutId = window.setTimeout(execute, ms)
    // 返回清理函数
    return () => {
        isRunning = false
        if (timeoutId !== undefined) {
            window.clearTimeout(timeoutId)
            timeoutId = undefined
        }
    }
}