本文写于 2022 年 5 月 13 日
线程池在多线程语言中,我们通常不会随意的在需要启动线程的时候去启动,而是会选择创建一个线程池。
所谓线程池,本意其实就是(不止这些作用,其余作用可以自行查阅):
- 节省操作系统资源
- 限制最大线程数。
对于 JavaScript 来说,虽然不存在“启动线程”这种问题,但我们还是可以通过类似的思想,来限制我们做异步操作的数量。
例如我们同时将进行 1000 个异步请求,那么我们可以约束同时执行的最大异步操作数为 100。
第一次我们执行 100 个异步任务,还剩余 900 个未执行。接着每当前面的异步任务有执行完毕之后,我们就会将空出来的位置执行下一个异步任务。
分析首先我们需要一个数组,用它来存储尚未执行的任务,每个任务都是一个函数,这个函数必须要返回一个 Promise。
type Task = () => Promise<unknown>;
const tasks: Task[] = [];
其次我们需要一个方法来进行任务的添加。
function addTask(task: Task): void;
最后我们需要一个函数来执行我们所有的 task。
而在这之前,我们还需要定义一个值,来定义同时执行的异步任务的最大数量。
function execTasks(): void;
实现
根据我们的分析,我们可以写下基础的代码如下:
interface TaskPool {
addTask(task: Task): void;
}
type Task = () => Promise<unknown>;
function newTaskPool(max = 10): TaskPool {
const tasks: Task[] = [];
function addTask(task: Task): void {}
function execTasks(): void {}
}
新增任务非常简单,我们写出如下代码填充 addTask
。
function addTask(task: Task): void {
tasks.push(task);
}
接下来就是重头戏。如何实现 execTasks
方法来限制最大异步任务数量呢?
首先我们来明确一点,在下面这个场景中,如果 foo
函数是异步操作,那么是不会阻塞我们的代码执行的。
console.log("Before");
foo();
console.log("After");
那么我们可以这么操作:
- 定义一个变量用来记录当前的空闲任务数量;
- 执行
execTasks
时,会选取当前任务数量和空闲任务数二者相比较小的一个; - 根据该值进行循环,每次循环弹出
tasks
第一位的任务进行执行; - 执行前将空闲任务数 -1,执行完毕后空闲任务数 +1,并再次执行
execTasks
。
let leisure = max;
function execTasks(): void {
if (tasks.length === 0) return;
const execTaskNum = Math.min(tasks.length, leisure);
for (let i = 0; i < execTaskNum; i++) {
const task = tasks.shift();
if (!task) continue;
leisure--;
task().finally(() => {
leisure++;
execTasks();
});
}
}
最后我们只剩下了一个问题了,我们如何在 addTask
后执行 execTasks
,但又不会让下面这种情况导致频繁执行 execTasks
:
for (let i = 0; i < 100; i++) addTask();
可以利用防抖 + setTimeout(() => {},0)
的特性来完成。
function addTask(task: Task) {
tasks.push(task);
execTasksAfterAdd();
}
// 这里借用了 lodash 的 debounce 函数,具体实现不多说,可以看我以前的文章:防抖与节流
const execTasksAfterAdd = debounce(execTasks);
完整代码:
import { debounce } from "lodash";
interface TaskQueue {
addTask: (task: () => Promise<any>) => void;
}
function newTaskQueue(maxTaskNum = 10): TaskQueue {
let _leisure = maxTaskNum;
const _tasks: Array<() => Promise<any>> = [];
function addTask(task: () => Promise<any>) {
_tasks.push(task);
execAfterTask();
}
const execAfterTask = debounce(execTasks);
function execTasks() {
if (_tasks.length === 0) return;
const execTaskNum = Math.min(_tasks.length, _leisure);
for (let i = 0; i < execTaskNum; i++) {
const task = _tasks.shift();
if (!task) continue;
_leisure--;
task().finally(() => {
_leisure++;
execTasks();
});
}
}
return { addTask };
}
const queue = newTaskQueue(5);
for (let i = 0; i < 10; i++) {
queue.addTask(function () {
return new Promise<void>((resolve) => {
setTimeout(() => resolve(), 800);
});
});
}
使用场景
其实这种做法的使用场景是比较少的。
绝大多数情况我们都不需要这么去做,除非碰到很极端的需求。
例如我们需要用 Node.js 去设计一个吞吐量极大的服务,那么同时发生大量的网络请求很可能把带宽直接打满,导致后续的请求无法打到该服务,此时就可以使用任务池来控制最大网络请求量。
(完)