一、前提介绍

我在前文实现过一个【批量任务调度管理器】的 demo，能实现简单的任务批量并发分组，过滤等操作。但是还有很多优化空间，所以查找一些优化的库，

主要想优化两个方面， 上篇提到的：

针对 3，其实可以自己手写一个，也可以依靠 如什么来实现。

针对 2，最难的是根据【当前系统负载或网络状况调整并发请求数量】，这必须需要引入一个检测系统的东西。

而动态调整数量，我现在的代码其实已经可以满足，我只是无法【获取当前系统负载或网络状况】。

当然我搜了一些，也有推荐 P-Queue 这类队列库来现在我现在的功能的，它有一些优点，比如：<code>动态调整并发数、任务优先级、延迟、减少手动实现的复杂性，简化代码逻辑，提高可读性和维护性等等。

最后实现功能：基于 P-Queue 和 AbortController 的批量操作功能，其中包括了动态并发控制、网络状态监测和请求取消功能。

注意：

AbortController 是 Web API 的一部分，可以用来中止一个或多个 fetch 请求。虽然 Axios 默认 不支持 直接使用 AbortController，但可以通过传递 AbortController 的 signal 属性到 Axios请求 中来实现这一功能。对于循环请求，为每个请求创建一个新的 AbortController 实例，并在合适的时机调用 abort 方法就可以实现中止

二、P-Queue 介绍

P-Queue 是一个基于 Promise 的优先队列，它允许你控制并发任务的数量，并且可以设置任务的优先级。P-Queue 的使用非常简单，只需要创建一个队列实例，然后使用 add 方法添加任务即可。

优点：

1. 固定并发数: 创建 P-Queue 实例时指定一个并发数，例如 5。这表示队列最多同时处理 5 个任务。

2. 自动调度：只需要将任务添加到队列中，而不需要关心任务的调度和执行顺序。P-Queue 会根据你指定的并发数自动处理任务——我现在自己手写的那个只是考虑使用 for 循环启动多个并发请求，并通过递归调用来处理下一任务。这种方法不涉及任务队列的概念，而是直接处理任务并发

3. 任务执行：任务被添加到队列时，P-Queue 会立即检查当前的并发任务数量。如果当前执行的任务数少于并发数限制，它会立即开始执行新的任务。

如果当前正在执行的任务数达到并发限制，P-Queue 会将新任务放入队列中，等待直到有空闲的并发槽位（即当前正在运行的任务数减少到低于并发限制）。

4. 任务完成：当一个任务完成时，P-Queue 会检查队列中的其他待处理任务。如果有任务在队列中等待并且当前并发数仍然未达到限制，它会自动开始执行下一个任务。

使用案例：

import PQueue from 'p-queue'

import axios from 'axios'

// 创建一个队列，设置并发数为 5

const queue = new PQueue({ -- --> concurrency: 5 })

const tasks = [

() => axios.get('/api/task1'),

() => axios.get('/api/task2')

// 更多任务...

]

// 添加任务到队列中

tasks.forEach((task) => {

queue.add(task)

})

// 动态调整并发数——可以根据系统的情况来调整

setTimeout(() => {

queue.concurrency = 10 // 动态调整并发数

}, 5000)

// 等待所有任务完成

queue.onIdle().then(() => {

console.log('All tasks completed')

})

三、动态检测的实现

我曾想过使用一些第三方库来实现检测当前系统的情况，然后动态控制并发数量，这样会更友好，查了一些方案，感觉没有必要。

实际的负载检测逻辑通常依赖于系统性能指标和资源使用情况。一些常见的负载检测方法和第三方库:

os-utils：提供系统负载和资源使用情况的简单接口。systeminformation：提供详细的系统信息，包括 CPU、内存、网络等。node-os-utils：获取系统的 CPU、内存、磁盘和网络信息，适合用于 Node.js 环境。

对于大多数应用，特别是当任务负载和并发需求相对稳定时，动态调整可能显得过于复杂。遂放弃。

四、优化效果

最后，我的方案如下：

基于 P-Queue 和 AbortController 的批量操作功能，其中包括了动态并发控制、网络状态监测和请求取消功能。以下是实现功能的详细描述：

功能优化概述

动态并发控制：getConcurrency：根据设备的硬件线程数（CPU 核心数）动态计算并发请求数量，一个比较基础的检测，也可以考虑其他的检查队列 queue：使用 P-Queue 创建任务队列，初始化并发数量。监听网络是否中断，再批量请求处理：每个请求的处理函数，使用 AbortController 允许在网络断开时中断请求。注意：组件挂载时初始化网络状态监听、也可以考虑监听网络的情况更新 P-Queue 的并发数量

优点是：可扩展性强，还是比较简单灵活，结合了并发控制、请求取消和网络状态检测。能够提升性能，优化用户体验。

代码实现

import PQueue from 'p-queue'

import axios from 'axios'

onMounted(() => {

window.addEventListener('online', handleNetworkChange)

window.addEventListener('offline', handleNetworkChange)

})

// 创建 AbortController 实例

let abortController = new AbortController()

// 网络状态检测函数

const isOnline = () => navigator.onLine

// 定义动态并发控制函数（根据需要自定义）

const getConcurrency = () => {

//目的是根据系统的硬件资源或其他条件动态调整并发请求的数量。具体来说，它通过 navigator.hardwareConcurrency 获取设备的硬件线程数（CPU核心数），然后将其限制在一个合适的范围内（在这个例子中是 1 到 10)

// 你可以根据系统负载、网络状况等动态调整并发数

return Math.max(1, Math.min(10, navigator.hardwareConcurrency || 4))

}

// 创建 P-Queue 实例

const queue = new PQueue({ concurrency: getConcurrency() })

// 更新并发数的函数

const updateConcurrency = () => {

queue.concurrency = getConcurrency()

}

// 存储每个请求的 AbortController 实例

const controllers = ref([])

const batchOperation = (title, operationType, axiosConfig, shouldFilterList) => {

const successCount = ref(0)

const errorCount = ref(0)

// 执行批量操作

startLoading()

const requestTaskList = shouldFilterList

? tableData.value.filter((item, index, arr) => arr.findIndex((val) => val.id === item.id) === index)

: tableData.value

if (requestTaskList.length === 0) {

ElMessage.error('没有可操作的任务！')

stopLoading()

return

}

// 定义请求函数

const requestFunction = async (row) => {

// 创建新的 AbortController 实例

const controller = new AbortController()

controllers.value.push(controller) // 将控制器添加到集合中

const params = { }

try {

const res = await axios.request({

url: axiosConfig.url,

method: axiosConfig.method,

params: params,

signal: abortController.signal // 使用 AbortController 实例

})

if (res.data.code === 200) {

successCount.value++

} else {

errorCount.value++

}

} catch (error) {

if (error.name === 'AbortError') {

console.warn('Request was aborted due to network being offline.')

} else {

errorCount.value++

console.error('Request failed:', error)

}

}

}

requestTaskList.forEach((row) => {

queue.add(() => requestFunction(row))

})

queue.onIdle().then(() => {

stopLoading(tabs)

ElNotification({

title: `${ title}结果`,

message: `共 ${ requestTaskList.length} 个任务，${ successCount.value} 个处理成功，${ errorCount.value} 个处理失败。`,

type: errorCount.value === 0 ? 'success' : 'warning'

})

controllers.value = [] // 清空控制器集合

})

// 动态调整并发数的定时器（每隔 5 分钟更新一次）——感觉作用不大，会导致不必要的性能开销

// setInterval(() => {

// queue.concurrency = getConcurrency()

// }, 300000)

}

// 取消所有请求

const cancelAllRequests = () => {

controllers.value.forEach((controller) => {

controller.abort()

})

controllers.value = [] // 清空控制器集合

}

// 监听网络状态变化

const handleNetworkChange = () => {

if (!isOnline()) {

console.log('Network is offline. Cancelling ongoing requests.')

cancelAllRequests() // 取消所有请求

} else {

console.log('Network is back online. Resuming requests.')

abortController = new AbortController() // 创建新的 AbortController 实例

}

}

// 在组件卸载之前，取消所有未完成的请求

onBeforeUnmount(() => {

isUnmount.value = true

window.removeEventListener('online', handleNetworkChange)

window.removeEventListener('offline', handleNetworkChange)

cancelAllRequests() // 组件卸载时取消所有请求

})

效果评价

1.并发数测试

同样 20 条，优化前：

优化第一版，上篇：

优化第二版，这版：

其实吧，我也看不出来，性能有啥变化，累了，还是太少了吧……

但是，代码友好了一些些吧，并发上来讲，比之前好很多。

2.取消请求测试

同样 20 条，优化前，没有加取消的功能，断网会——500！如果你的 axios 拦截里对 500 进行处理的话，就会误伤。

优化后：

断网前发送的请求会 500，但是断网后，有一些请求还未发出去，所以会被取消。

效果好一点点。

五、总结

对上面的功能主要做了队列并发处理，简单的动态检测系统网络是否连接然后中断请求的处理，算是优化了该 demo。以后如果接触到更好的办法会继续更新！