代码之家 › 专栏 › 技术社区 › lanetrotro

限制节点中对cassandra db的并行请求数

promise cassandra asynchronous node.js

lanetrotro · 技术社区 · 7 年前

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

问题是,我被限制在2048个并行请求。

这是我编的代码:

索引。

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)
});

insertdb.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秒添加一个wait函数,在for循环中设置一个timout(但是它不起作用,因为我已经有了承诺),我也尝试了 p-queue 和 p-limit 但它似乎也不起作用。

我有点困在这里,我想我错过了一些琐碎的东西,但我真的不明白。

谢谢

2 回复 | 直到 7 年前

jorgebg 7 年前

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

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

下面是一个关于如何在循环中处理项时限制并发执行量的示例: 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);
  }
}

lanetrotro 7 年前

好吧,所以我找到了一个解决方法来达到我的目标。我把所有的疑问都写进了一个文件

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

然后我用了

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

在Cassandra中插入我的所有查询