What is concurrency control? How to implement concurrency control in JavaScript?

In the daily development process, you may encounter concurrency control scenarios, such as controlling the number of concurrent requests. So how to implement concurrency control in JavaScript? Before answering this question, let's briefly introduce concurrency control.

Suppose there are 6 to-do tasks to be executed, and we want to limit the number of tasks that can be executed at the same time, that is, at most 2 tasks can be executed at the same time. When any task inExecuting Task Listis completed, the program will automatically obtain a new to-do task fromTo-Do Task Listand add the task toExecuting task list. In order to allow everyone to understand the above process more intuitively, Abago specially drew the following 3 pictures:

1.1 Stage 1

1.2 Phase Two

1.3 Phase Three

Okay, after introducing concurrency control, Brother Abao will use theasync-poollibrary on Github to introduce the specific implementation of asynchronous task concurrency control.

async-pool：https://github.com/rxaviers/async-pool

Run multiple promise-returning & async functions with limited concurrency using native ES6/ ES7.

2. Implementation of concurrency control

async-pool This library provides two different versions of implementation, ES7 and ES6. After analyzing the Before the specific implementation, let's take a look at how to use it.

2.1 The use of asyncPool

const timeout = i => new Promise(resolve => setTimeout(() => resolve(i), i)); await asyncPool(2, [1000, 5000, 3000, 2000], timeout);

In the above code, we use theasync-poolprovided by this libraryasyncPoolFunction to implement concurrency control of asynchronous tasks.asyncPoolThe signature of the function is as follows:

function asyncPool(poolLimit, array, iteratorFn){ ... }

The function receives 3 parameters:

• ##poolLimit(numeric type): represents The number of limited concurrency;
• array(array type): represents the task array;
• iteratorFn(function type): represents the iteration function, used To implement processing of each task item, this function will return a Promise object or an asynchronous function.

For the above example, after using the

asyncPoolfunction, the corresponding execution process is as follows:

const timeout = i => new Promise(resolve => setTimeout(() => resolve(i), i)); await asyncPool(2, [1000, 5000, 3000, 2000], timeout); // Call iterator (i = 1000) // Call iterator (i = 5000) // Pool limit of 2 reached, wait for the quicker one to complete... // 1000 finishes // Call iterator (i = 3000) // Pool limit of 2 reached, wait for the quicker one to complete... // 3000 finishes // Call iterator (i = 2000) // Itaration is complete, wait until running ones complete... // 5000 finishes // 2000 finishes // Resolves, results are passed in given array order `[1000, 5000, 3000, 2000]`.

By observing the above annotation information, We can roughly understand the control flow inside the

asyncPoolfunction. Let's first analyze the ES7 implementation of theasyncPoolfunction.

Follow "Full Stack Road to Immortality" and read 4 free e-books originally created by Brother A Bao (a total of 30,000 downloads) and more than 50 TS series tutorials.

2.2 asyncPool ES7 implementation

async function asyncPool(poolLimit, array, iteratorFn) { const ret = []; // 存储所有的异步任务 const executing = []; // 存储正在执行的异步任务 for (const item of array) { // 调用iteratorFn函数创建异步任务 const p = Promise.resolve().then(() => iteratorFn(item, array)); ret.push(p); // 保存新的异步任务 // 当poolLimit值小于或等于总任务个数时，进行并发控制 if (poolLimit <= array.length) { // 当任务完成后，从正在执行的任务数组中移除已完成的任务 const e = p.then(() => executing.splice(executing.indexOf(e), 1)); executing.push(e); // 保存正在执行的异步任务 if (executing.length >= poolLimit) { await Promise.race(executing); // 等待较快的任务执行完成 } } } return Promise.all(ret); }

In the above code, the features of

Promise.allandPromise.raceare fully utilized. Combined with theasync awaitfeature provided in ES7, the concurrency control function is finally realized. Using theawait Promise.race(executing);line of statement, we will wait for the faster task in theexecuting task listto complete before continuing to execute the next loop.

asyncPool ES7 implementation is relatively simple. Next, let’s take a look at how to achieve the same function without using the

async awaitfeature.

2.3 asyncPool ES6 implementation

function asyncPool(poolLimit, array, iteratorFn) { let i = 0; const ret = []; // 存储所有的异步任务 const executing = []; // 存储正在执行的异步任务 const enqueue = function () { if (i === array.length) { return Promise.resolve(); } const item = array[i++]; // 获取新的任务项 const p = Promise.resolve().then(() => iteratorFn(item, array)); ret.push(p); let r = Promise.resolve(); // 当poolLimit值小于或等于总任务个数时，进行并发控制 if (poolLimit <= array.length) { // 当任务完成后，从正在执行的任务数组中移除已完成的任务 const e = p.then(() => executing.splice(executing.indexOf(e), 1)); executing.push(e); if (executing.length >= poolLimit) { r = Promise.race(executing); } } // 正在执行任务列表 中较快的任务执行完成之后，才会从array数组中获取新的待办任务 return r.then(() => enqueue()); }; return enqueue().then(() => Promise.all(ret)); }

In the ES6 implementation version, the core control logic is implemented through the internally encapsulated

enqueuefunction. When thePromiseobject returned byPromise.race(executing)becomes completed, theenqueuefunction will be called, fromarrayGet new to-do tasks from the array.

3. Brother A Bao has something to say

In the specific implementation of ES7 and ES6 of

asyncPool, we have usedPromise.allandPromise.racefunctions. Among them, handwrittenPromise.allis a common interview question. Just taking advantage of this opportunity, Brother A Bao and everyone came to handwrite simple versions of thePromise.allandPromise.racefunctions.

3.1 手写 Promise.all

Promise.all(iterable)方法会返回一个 promise 对象，当输入的所有 promise 对象的状态都变成resolved时，返回的 promise 对象就会以数组的形式，返回每个 promise 对象 resolve 后的结果。当输入的任何一个 promise 对象状态变成rejected时，则返回的 promise 对象会 reject 对应的错误信息。

Promise.all = function (iterators) { return new Promise((resolve, reject) => { if (!iterators || iterators.length === 0) { resolve([]); } else { let count = 0; // 计数器，用于判断所有任务是否执行完成 let result = []; // 结果数组 for (let i = 0; i < iterators.length; i++) { // 考虑到iterators[i]可能是普通对象，则统一包装为Promise对象 Promise.resolve(iterators[i]).then( (data) => { result[i] = data; // 按顺序保存对应的结果 // 当所有任务都执行完成后，再统一返回结果 if (++count === iterators.length) { resolve(result); } }, (err) => { reject(err); // 任何一个Promise对象执行失败，则调用reject()方法 return; } ); } } }); };

需要注意的是对于Promise.all的标准实现来说，它的参数是一个可迭代对象，比如 Array、String 或 Set 等。

3.2 手写 Promise.race

Promise.race(iterable)方法会返回一个 promise 对象，一旦迭代器中的某个 promise 对象resolved或rejected，返回的 promise 对象就会 resolve 或 reject 相应的值。

Promise.race = function (iterators) { return new Promise((resolve, reject) => { for (const iter of iterators) { Promise.resolve(iter) .then((res) => { resolve(res); }) .catch((e) => { reject(e); }); } }); };

本文阿宝哥带大家详细分析了async-pool异步任务并发控制的具体实现，同时为了让大家能够更好地理解async-pool的核心代码。最后阿宝哥还带大家一起手写简易版的Promise.all和Promise.race函数。其实除了Promise.all函数之外，还存在另一个函数 ——Promise.allSettled，该函数用于解决Promise.all存在的问题，感兴趣的小伙伴可以自行研究一下。

四、参考资源

• Github - async-pool
• MDN - Promise.all
• MDN - Promise.race
• MDN - Promise.allSettled

更多编程相关知识，请访问：编程视频！！

The above is the detailed content of What is concurrency control? How to implement concurrency control in JavaScript?. For more information, please follow other related articles on the PHP Chinese website!