Блог

Я использую асинхронную очередь для обработки огромного объема данных. Очередь отлично работает, пока я не попытаюсь выполнить обновление в базе данных с помощью метода MongoDB findOneAndUpdate.

Сначала я устанавливаю очередь и начинаю передавать в нее объекты из потока узлов:

 //Create queue to process all items
let q = async.queue(processLink, 2);

// Create Node Stream
let createStream = function () {
    let stream = fs.createReadStream(LinkData, {encoding: 'utf8'});
    let parser = JSONStream.parse('RECDATA.*');
    return stream.pipe(parser);
};

//Listen to 'data' event on stream and add object to queue
createStream().on('data', function(link){
    q.push(link)
});
  

Вот моя функция задачи, ‘processLink’. Именно здесь у меня возникают проблемы с отслеживанием проблемы. Всякий раз, когда запускается обратный вызов findOneAndUpdate, он входит в один из условных блоков, и я получаю сообщение, отправленное на консоль, но когда я вызываю асинхронный обратный вызов processComplete() , задача завершается не так, как ожидалось.

Как следует из названия, почему мои асинхронные обратные вызовы не завершают каждую задачу?

 function processLink(link, processComplete){
    if(_.includes(link.URL, 'www.usda.gov') amp;amp; _.includes(link.URL, '?recid=')){
        let url_items = _.split(link.URL, '=',2);
        let facilityOrgID = url_items[1];
        let update = {$push: {"links": link}};

        if(_.isNumber(parseInt(facilityOrgID)) amp;amp; facilityOrgID.length > 4 ){

            Facility.findOneAndUpdate({facilityOrgID: parseInt(facilityOrgID)}, update, (err, result) => {

                if(err !== null){
                    console.log("Error:",err);
                    return processComplete(err); /** NOT FIRING **/
                } else if(err === null amp;amp; result !== null){

                    console.log("Link added to:", result.name);
                    return processComplete(); /** NOT FIRING **/ 
                }else if(err === null amp;amp; result === null){

                    console.log('Facility not in database');
                    processComplete(); /** NOT FIRING **/ 
                }else{
                    console.log('Something has gone terrible wrong');
                }
            });

        }else{
            console.log("Invalid facilityID");
            return processComplete();
        }

    }else{
        console.log('Link Discarded:', link.URL);
        processComplete(); /** Fires normally **/ 
    }
}
  

Я решил свою проблему этим утром. Обратные вызовы были в порядке, но скорее было условное состояние данных, о котором я не знал, таким образом, это приводило к состоянию, в котором код никогда не вызывал бы обратный вызов processComplete() .

Если кто-то еще находится в подобной привязке, и у вас есть тонны данных, я бы позаботился о том, чтобы ваши данные сообщались точно, и не было каких-либо крайних случаев, которые обрабатывались не так, как ожидалось.