Typescript/ES7 中异步/等待的有限并行性 [英] Limited parallelism with async/await in Typescript/ES7

问题描述

我一直在尝试使用 Typescript，但现在我对如何有效地使用 async/await 有点困惑.

I've been experimenting a bit with Typescript, but I'm now a bit stuck on how to use async/await effectively.

我正在向数据库中插入一堆记录，我需要获取每次插入返回的 ID 列表.下面的简化示例在一般情况下有效，但它并不像我想要的那么优雅，而且它是完全顺序的.

I'm inserting a bunch of records into a database, and I need to get the list of IDs that are returned by each insert. The following simplified example works in general, but it is not quite as elegant as I'd like and it is completely sequential.

async function generatePersons() {
    const names = generateNames(firstNames, lastNames);
    let ids = []
    for (let name of names) {
        const id = await db("persons").insert({
            first_name: name.firstName,
            last_name: name.lastName,
        }).returning('id');
        ids.push(id[0])
    }
    return ids
}

我尝试使用 map 来避免手动创建 ids 列表，但我可以让它工作.

I tried to use map to avoid creating the ids list manually, but I could get this to work.

我还想拥有有限数量的并行性.所以我的异步调用应该并行发生到一定的限制，例如我只希望有 10 个开放请求，但不会更多.

What I'd also like to have is a limited amount of parallelism. So my asynchronous calls should happen in parallel up to a certain limit, e.g. I'd only ever like to have 10 open requests, but not more.

在 Typescript 或 Javascript ES7 中使用 async/await 实现这种有限的并行性是否有一种相当优雅的方法?还是我试图让这个功能做一些它不适合做的事情?

Is there a reasonably elegant way of achieving this kind of limited parallelism with async/await in Typescript or Javascript ES7? Or am I trying to get this feature to do something it was not intended for?

_{PS:我知道数据库有批量插入方法，这个例子有点人为，因为我可以使用这些方法来解决这个特定问题.但这让我想知道我没有可用的预定义批量方法的一般情况，例如带有网络请求}

推荐答案

Promise.all 将允许您等待所有请求停止完成，而不会阻止它们的创建.

Promise.all will allow you to wait for all requests to stop finishing, without blocking their creation.

然而，它确实听起来你有时想阻止.具体来说，听起来您想限制在任何给定时间进行中的请求数量.这是我提出的一些东西(但还没有完全测试！)

However, it does sound like you want to block sometimes. Specifically, it sounded like you wanted to throttle the number of requests in flight at any given time. Here's something I whipped up (but haven't fully tested!)

async function asyncThrottledMap<T, U>(maxCount: number, array: T[], f: (x: T) => Promise<U>) {
    let inFlight = new Set<Promise<U>>();
    const result: Promise<U>[] = [];

    // Sequentially add a Promise for each operation.
    for (let elem of array) {

        // Wait for any one of the promises to complete if there are too many running.
        if (inFlight.size >= maxCount) {
            await Promise.race(inFlight);
        }

        // This is the Promise that the user originally passed us back.
        const origPromise = f(elem);
        // This is a Promise that adds/removes from the set of in-flight promises.
        const handledPromise = wrap(origPromise);
        result.push(handledPromise);
    }

    return Promise.all(result);

    async function wrap(p: Promise<U>) {
        inFlight.add(p);
        const result = await p;
        inFlight.delete(p);
        return result;
    }
}

上面，inFlight 是一组当前正在发生的操作.

Above, inFlight is a set of operations that are currently taking place.

result 是一个 wrapped Promise 的数组.这些包装好的承诺中的每一个基本上都是从 inFlight 操作集中添加或删除操作.如果进行中的操作太多，则使用 Promise.race 来完成任何一个进行中的操作.

The result is an array of wrapped Promises. Each of those wrapped promises basically adds or removes operations from the set of inFlight operations. If there are too many in-flight operations, then this uses Promise.race for any one of the in-flight operations to complete.

希望有帮助.

这篇关于Typescript/ES7 中异步/等待的有限并行性的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！