读取具有 settimeout 最大值的流已达到错误

Question

我正在尝试读取一些大型 CSV 文件并处理这些数据，因此处理中存在速率限制，因此我想在每个请求之间添加 1mnt 延迟。我尝试了设置超时，但最后，知道设置超时有限制并得到以下错误。我不确定任何其他方式来处理这种情况，CSV 文件有超过 1M 的记录。我在这里做错什么了吗？

错误

超时持续时间设置为 1。（节点：41） TimeoutOverflowWarning： 2241362000 不适合 32 位有符号整数。

示例代码：

   const Queue = require('bull');
const domainQueue = new Queue(config.api.crawlerQ, {
  redis: connectRedis(),
});
let ctr = 0;
function processCSV (name, fileName, options)  {
  return new Promise((resolve, reject) => {
    console.log('process csv started', new Date());
    let filePath = config.api.basePath + fileName;
    stream = fs.createReadStream(filePath)
        .on('error', (error) => {
          // handle error
          console.log('error processing csv');
          reject(error);
        })
        .pipe(csv())
        .on('data', async (row) => {
          ctr++
          increment(row, ctr)
        })
        .on('end', () => {
          console.log('stream processCSV end', fileName, new Date());
          resolve(filePath);
        })
  });

}

async function increment(raw, counter) {
  setTimeout(async function(){
    console.log('say i am inside a function', counter, new Date());
    domainQueue.add(data, options); // Add jobs to queue - Here i Need a delay say 1mnt, if i
    // add jobs without delay it will hit ratelimit 
  }, 60000 * counter);

}

function queueWorkerProcess(value) { // Process jobs in queue and save in text file 
  console.log('value', value, new Date());
  return new Promise(resolve => {
    resolve();
  });

}

Answer 1

这是一个总体思路。您需要跟踪正在处理的项目数量，以限制使用的内存量并控制存储结果的任何资源的负载。

当您达到飞行中的数量限制时，您会暂停流。当您回到限制以下时，您将恢复流。您在.add() 上增加一个计数器并在completed 消息上减少一个计数器以跟踪事情。您可以在此处暂停或恢复直播。

仅供参考，仅在某处插入 setTimeout() 对您没有帮助。为了控制内存使用，一旦处理中的项目过多，您必须暂停流中的数据流。然后，当项目恢复到阈值以下时，您可以恢复流。

这是一个大概的样子：

const Queue = require('bull');
const domainQueue = new Queue(config.api.crawlerQ, {
    redis: connectRedis(),
});

// counter that keeps track of how many items in the queue
let queueCntr = 0;

// you tune this constant up or down to manage memory usage or tweak performance
// this is what keeps you from having too many requests going at once
const queueMax = 20;

function processCSV(name, fileName, options) {
    return new Promise((resolve, reject) => {
        let paused = false;

        console.log('process csv started', new Date());
        const filePath = config.api.basePath + fileName;

        const stream = fs.createReadStream(filePath)
            .on('error', (error) => {
                // handle error
                console.log('error processing csv');
                domainQueue.off('completed', completed);
                reject(error);
            }).pipe(csv())
            .on('data', async (row) => {
                increment(row, ctr);
                if (queueCntr)
            })
            .on('end', () => {
                console.log('stream processCSV end', fileName, new Date());
                domainQueue.off('completed', completed);
                resolve(filePath);
            });

        function completed() {
            --queueCntr;
            // see if queue got small enough we now resume the stream
            if (paused && queueCntr < queueMax) {
                stream.resume();
                paused = false;
            }
        }

        domainQueue.on('completed', completed);

        function increment(raw, counter) {
            ++queueCntr;
            domainQueue.add(data, options);
            if (!paused && queueCntr > queueMax) {
                stream.pause();
                paused = true;
            }
        }
    });
}

而且，如果您使用不同的文件多次调用 processCSV()，您应该对它们进行排序，以便在第一个完成之前不要调用第二个，在第二个完成之前不要调用第三个已完成，依此类推...您没有显示该代码，因此我们无法对此提出具体建议。