我目前正在解析一个文件并获取其数据,以便将它们推送到我的数据库中。为此,我创建了一个查询数组,并通过循环执行它们。
问题是我的并行请求数被限制为 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
但似乎也不起作用。
我有点卡在这里,我想我错过了一些微不足道的东西,但我真的不明白是什么。
谢谢
最佳答案
并行提交多个请求时(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);
}
}
关于node.js - 限制nodejs中cassandra db的并行请求数量,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/53390876/