在nodejs中限制对cassandra db的并行请求数量答案

【问题标题】：limiting number of parallel request to cassandra db in nodejs在nodejs中限制对cassandra db的并行请求数量
【发布时间】：2019-04-22 18:31:15
【问题描述】：

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

问题是我被限制为 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 循环中设置一个超时（但它不起作用，因为我已经在承诺中），我还尝试了 p-queue 和 @ 987654324@ 但它似乎也不起作用。

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

谢谢

【问题讨论】：

标签： node.js asynchronous cassandra promise

【解决方案1】：

好的，所以我找到了一种解决方法来实现我的目标。我将所有查询都写在一个文件中

const fs = require('fs')
fs.appendFileSync('my_file.cql', queries[i] + "\n");

然后我用了

child_process.exec("cqls --file my_file", function(err, stdout, stderr){})"

在 cassandra 中插入我的所有查询

【讨论】：

【解决方案2】：

当并行提交多个请求时（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);
  }
}

【讨论】：