我正在嘗試并行執行一些異步任務，同時限制同時運行的任務的最大數量。
有一個我想要實作的例子：
目前這項任務正在一個接一個地運行。它是這樣實作的：
export function signData(dataItem) {
cadesplugin.async_spawn(async function* (args) {
//... nestedArgs assignment logic ...
for (const id of dataItem.identifiers) {
yield* idHandler(dataItem, id, args, nestedArgs);
}
// some extra logic after all tasks were finished
}, firstArg, secondArg);
}
async function* idHandler(edsItem, researchId, args, nestedArgs) {
...
let oDocumentNameAttr = yield cadesplugin.CreateObjectAsync("CADESCOM.CPAttribute");
yield oDocumentNameAttr.propset_Value("Document Name");
...
// this function mutates some external data, making API calls and returns void
}
不幸的是，我無法對cadesplugin.*函式進行任何更改，但我可以Promise在我的代碼中使用任何外部庫（或內置庫）。
我在異步庫中找到了一些可能對我有用的方法（eachLimit和parallelLimit）以及一個顯示如何處理它的答案。
但是還有兩個問題我無法解決：
- 如何將主要引數傳遞給嵌套函式？
- main 函式是一個生成器函式，所以我仍然需要在 main 函式和嵌套函式中使用yield運算式
有一個指向cadesplugin.* 源代碼的鏈接，您可以在其中找到在我的代碼中使用的async_spawn（和另一個cadesplugin.*）函式。
這是我沒有運氣嘗試的代碼：
await forEachLimit(dataItem.identifiers, 5, yield* async function* (researchId, callback) {
//... nested function code
});
它導致Object is not async iterable錯誤。
另一種嘗試：
let functionArray = [];
dataItem.identifiers.forEach(researchId => {
functionArray.push(researchIdHandler(dataItem, id, args, nestedArgs))
});
await parallelLimit(functionArray, 5);
它什么都不做。
我以某種方式解決了這個問題，或者生成器功能不允許我這樣做？
uj5u.com熱心網友回復：
方釘，圓孔
您不能對這個問題使用異步迭代。串聯for await .. of運行是本性。阻塞并且回圈不會繼續，直到等待的承諾完成。您需要更精確的控制級別，您可以在其中強制執行這些特定要求。await
首先，我們有一個模擬myJob長時間計算的模擬。這很可能是對您應用中某些 API 的網路請求 -
// any asynchronous task
const myJob = x =>
sleep(rand(5000)).then(_ => x * 10)
使用此問答中的Pool定義，我們實體化要運行的并發執行緒數在哪里-Pool(size=4)size
const pool = new Pool(4)
對于人體工程學，我run在Pool類中添加了一個方法，使其更容易包裝和運行作業 -
class Pool {
constructor (size) ...
open () ...
deferNow () ...
deferStacked () ...
// added method
async run (t) {
const close = await this.open()
return t().then(close)
}
}
現在我們需要撰寫一個使用我們pool運行的效果myJob。在這里，您還將決定如何處理結果。請注意，promise必須包含在 thunk 中，否則池無法控制它何時開始 -
async function myEffect(x) {
// run the job with the pool
const r = await pool.run(_ => myJob(x))
// do something with the result
const s = document.createTextNode(`${r}\n`)
document.body.appendChild(s)
// return a value, if you want
return r
}
現在通過映射myEffect您的輸入串列來運行所有內容。在我們的示例中myEffect，return r 這意味著在獲取所有結果后結果也可用。這是可選的，但演示了程式如何知道一切何時完成 -
Promise.all([1,2,3,4,5,6,7,8,9,10,11,12].map(myEffect))
.then(JSON.stringify)
.then(console.log, console.error)
完整的程式演示
在下面的功能演示中，我壓縮了定義，以便我們可以一次看到它們。運行程式以在您自己的瀏覽器中驗證結果 -
class Pool {
constructor (size = 4) { Object.assign(this, { pool: new Set, stack: [], size }) }
open () { return this.pool.size < this.size ? this.deferNow() : this.deferStacked() }
async run (t) { const close = await this.open(); return t().then(close) }
deferNow () { const [t, close] = thread(); const p = t.then(_ => this.pool.delete(p)).then(_ => this.stack.length && this.stack.pop().close()); this.pool.add(p); return close }
deferStacked () { const [t, close] = thread(); this.stack.push({ close }); return t.then(_ => this.deferNow()) }
}
const rand = x => Math.random() * x
const effect = f => x => (f(x), x)
const thread = close => [new Promise(r => { close = effect(r) }), close]
const sleep = ms => new Promise(r => setTimeout(r, ms))
const myJob = x =>
sleep(rand(5000)).then(_ => x * 10)
async function myEffect(x) {
const r = await pool.run(_ => myJob(x))
const s = document.createTextNode(`${r}\n`)
document.body.appendChild(s)
return r
}
const pool = new Pool(4)
Promise.all([1,2,3,4,5,6,7,8,9,10,11,12].map(myEffect))
.then(JSON.stringify)
.then(console.log, console.error)放慢速度
Pool以上盡可能快地運行并發作業。您可能還throttle對原帖中介紹的內容感興趣。Pool我們可以將我們的作業包裝起來，而不是變得更復雜，throttle讓呼叫者控制作業應該花費的最短時間 -
const throttle = (p, ms) =>
Promise.all([ p, sleep(ms) ]).then(([ value, _ ]) => value)
我們可以添加一個throttlein myEffect。現在，如果myJob運行速度非常快，在下一個作業運行之前至少會過去 5 秒 -
async function myEffect(x) {
const r = await pool.run(_ => throttle(myJob(x), 5000))
const s = document.createTextNode(`${r}\n`)
document.body.appendChild(s)
return r
}
轉載請註明出處，本文鏈接：https://www.uj5u.com/qianduan/392993.html
標籤：javascript 异步 异步等待 并行处理 异步.js