node.js - 限制nodejs中cassandra db的并行请求数量

Question

我目前正在解析一个文件并获取其数据，以便将它们推送到我的数据库中。为此，我创建了一个查询数组，并通过循环执行它们。

问题是我的并行请求数被限制为 2048 个。

这是我编写的代码:

index.js=>

const ImportClient = require("./scripts/import_client_leasing")
const InsertDb = require("./scripts/insertDb")

const cassandra = require('cassandra-driver');
const databaseConfig = require('./config/database.json');


const authProvider = new cassandra.auth.PlainTextAuthProvider(databaseConfig.cassandra.username, databaseConfig.cassandra.password);

const db = new cassandra.Client({
    contactPoints: databaseConfig.cassandra.contactPoints,
    authProvider: authProvider
});

ImportClient.clientLeasingImport().then(queries => { // this function parse the data and return an array of query
    return InsertDb.Clients(db, queries);    //inserting in the database returns something when all the promises are done
}).then(result => {
    return db.shutdown(function (err, result) {});
}).then(result => {
    console.log(result);
}).catch(error => {
    console.log(error)
});

插入Db.js =>

module.exports = {
    Clients: function (db, queries) {
        DB = db;
        return insertClients(queries);
    }
}

function insertClients(queries) {
    return new Promise((resolve, reject) => {
        let promisesArray = [];

        for (let i = 0; i < queries.length; i++) {
            promisesArray.push(new Promise(function (resolve, reject) {
                DB.execute(queries[i], function (err, result) {
                    if (err) {
                        reject(err)
                    } else {
                        resolve("success");
                    }
                });
            }));
        }
        Promise.all(promisesArray).then((result) => {
            resolve("success");
        }).catch((error) => {
            resolve("error");
        });
    });
}

我尝试了多种方法，例如添加一个等待函数，该函数每 x 秒在我的 for 循环中设置一个超时(但它不起作用，因为我已经在 promise 中)，我也尝试过 p-queue和p-limit但似乎也不起作用。

我有点卡在这里，我想我错过了一些微不足道的东西，但我真的不明白是什么。

谢谢

Answer 1

并行提交多个请求时(execute() 函数使用异步执行)，您最终会在不同级别之一排队:在驱动程序端、在网络堆栈或在服务器上边。过多的排队会影响完成每个操作所需的总时间。您应该随时限制同时请求的数量(也称为并发级别)，以获得高吞吐量和低延迟。

当考虑在代码中实现它时，您应该考虑启动固定数量的异步执行，使用并发级别作为上限，并且仅在该上限内的执行完成后才添加新操作。

以下是有关如何在处理循环中的项目时限制并发执行量的示例:https://github.com/datastax/nodejs-driver/blob/master/examples/concurrent-executions/execute-in-loop.js

简而言之:

// Launch in parallel n async operations (n being the concurrency level)
for (let i = 0; i < concurrencyLevel; i++) {
  promises[i] = executeOneAtATime();
}

// ...
async function executeOneAtATime() {
  // ...
  // Execute queries asynchronously in sequence
  while (counter++ < totalLength) {;
    await client.execute(query, params, options);
  }
}