проблема с несколькими асинхронными вызовами в node.js

Question

Я довольно новичок в node.js, так любезно прощаю мое невежество, что я пытаюсь сделать, это цикл через массив серверов и выделить максимум 60 задач каждому серверу , Однако количество входящих заданий может быть от 10 до 200.

Так что я хочу назначить 60 задач первому серверу, затем следующие 60 задач на второй сервер и так далее.

У меня есть следующая функция, которая, кажется, идут ужасно неправильно ..

function mount_the_tasks(server_info){ 
    async.forEach(server_info, function(single_server, callback1){ 
     Tasks.find({task_status : 0}).limit(single_server.can_perform_tasks).exec(function(err, tasks_for_server){ 
      async.forEach(tasks_for_server, function(single_task, callback2){ 
       Tasks.findOneAndUpdate({_id : single_task._id}, {task_status : 1, server_id : single_server.instance_id}, function(err, numberAffected){ 
        console.log(single_task.task_id+' -> '+single_server.instance_id); 
        callback2(); 
       }); 
      }, function(err){ 
       console.log('Moving to next server!'); 
       callback1(); 
      }); 
     }); 
    }, function(err){ 
     console.log('all done!'); 
    }); 
} 

Answer 1

Проблема вы собираетесь через все серверы параллельно. Что это означает для каждого сервера, вы назначаете первые 60 активных задач каждому, потому что к тому времени, когда будет запущен следующий сервер, предыдущий набор задач еще не выполнен, поэтому их статус еще равен 0. Чтобы исправить это , вам придется использовать async.eachSeries вместо этого, итерации по серверам, чтобы задачи были помечены как завершенные, прежде чем переходить на следующий сервер.

function mount_the_tasks(server_info){ 
    async.eachSeries(server_info, function(single_server, callback1){ 
     Tasks.find({task_status : 0}).limit(single_server.can_perform_tasks).exec(function(err, tasks_for_server){ 
      async.each(tasks_for_server, function(single_task, callback2){ 
       Tasks.findOneAndUpdate({_id : single_task._id}, {task_status : 1, server_id : single_server.instance_id}, function(err, numberAffected){ 
        console.log(single_task.task_id+' -> '+single_server.instance_id); 
        callback2(); 
       }); 
      }, function(err){ 
       console.log('Moving to next server!'); 
       callback1(); 
      }); 
     }); 
    }, function(err){ 
     console.log('all done!'); 
    }); 
} 

Другим вариантом было бы использовать .skip(), чтобы вызвать каждую итерацию, чтобы пропустить число задач, обрабатываемых предыдущим сервером. Это должно быть немного быстрее, чем предыдущий вариант, предполагая, что ваш сервер не является узким местом из-за количества одновременных отправки запросов.

function mount_the_tasks(server_info){ 
    var skip = 0; 
    async.each(server_info, function(single_server, callback1){ 
     Tasks.find({task_status : 0}).limit(single_server.can_perform_tasks).skip(skip).exec(function(err, tasks_for_server){ 
      async.each(tasks_for_server, function(single_task, callback2){ 
       Tasks.findOneAndUpdate({_id : single_task._id}, {task_status : 1, server_id : single_server.instance_id}, function(err, numberAffected){ 
        console.log(single_task.task_id+' -> '+single_server.instance_id); 
        callback2(); 
       }); 
      }, function(err){ 
       console.log('Moving to next server!'); 
       skip += single_server.can_perform_tasks; 
       callback1(); 
      }); 
     }); 
    }, function(err){ 
     console.log('all done!'); 
    }); 
}