Async.eachSeries：已调用回调

Question

这是一个有点长和凌乱的代码，但请耐心等待，因为我需要完成这个。

我正在尝试为每个用户更新一个 json 对象。我希望循环迭代等待异步过程结束以避免竞争条件。但是，这导致了回调地狱，现在我无法确定每个回调返回的正确位置。

我在 Nesting async.eachSeries 上提到了这个答案，并尝试根据它来构建我的代码。但它仍然不起作用。代码在 callback1() 处给出回调已经调用错误。

    async.eachOfSeries(res, function (value, camp, callback3) {
let _id = res[camp]._id;
let arr = res[camp].campaignID;
async.eachOfSeries(arr, function2, function (err) {
    callback3();
})


function function2(value1, i, callback2) {
    let users = arr[i].users;
    let id = arr[i].id;
    let loop = Math.ceil(users / 1000);
    let limit = 0,
        offset = 0;
    for (let j = 0; j < loop; j++) {
        if (users > 1000) {
            limit = 1000;
            users -= limit;
        } else {
            limit = users;
        }
        console.log(limit + " limit " + offset + " offset");
        var start = Date.now();
        while (Date.now() < start + 100) {}
        const request = mailjet
            .get("messagesentstatistics")
            .request({
                "CampaignID": id,
                "AllMessages": true,
                "Limit": limit,
                "Offset": offset
            })
        request
            .then((result) => {
                let data = result.body.Data;
                var loop = 0;
                async.eachOfSeries(data, function1, function (err) {
                    console.log("function");
                    callback2();
                })
                console.log("oooooo");
            })
            .catch((err) => {
                console.log(err);
            })
        offset += limit;
    }

    function function1(value2, val, callback1) {
        console.log(data +" data");
        let jsonObj = data[val];
        let email = jsonObj.ToEmail;
        jsonObj['retailer'] = res[camp].retailer;
        jsonObj['summary'] = 'f';
        let tempObj = {};
        tempObj[id] = jsonObj;
        let options = {
            new: true
        };
        let campId = id;
        User.addCampaignResponse(email, campId, tempObj, options, function (err, results) {
            if (err) {
                throw err;
            } else {
                console.log("aasd");
                Campaign.updateResponse(_id, function (err, results2) {
                    if (err)
                        throw err;
                    else {
                        console.log("asdasaadas");
                        callback1();
                    }
                }) // console.log(results);
            }
        })
    }

}

}, function (err) {
    callback(undefined, "doneeeeee");
})

还有比这更好的方法吗？我也可以在某个地方使用瀑布吗？我可以更改回调位置以避免错误吗？

---

编辑：简化代码

function function2(value1, i, callback2) {
    // ...
    const request = mailjet
                    .get("messagesentstatistics")
                    .request({
                       // ...
                    });
    request
       .then((result) => {
          // ...
          async.eachOfSeries(data, function1, function (err) {
            callback2();
          });
        })
        .catch((err) => {
          // ...
        });
    }

---

function function1(value2, val, callback1) {
  // ...
  User.addCampaignResponse(email, campId, tempObj, options, function (err, results) {
    if (err) {
      throw err;
    } else {
      Campaign.updateResponse(_id, function (err, results2) {
        if (err) throw err;
        else callback1();
      });
    }
  });
}

---

async.eachOfSeries(res, function (value, camp, callback3) {
    // ...

    async.eachOfSeries(arr, function2, function (err) {
      callback3();
    });

  },
  function (err) {
    callback(undefined, "doneeeeee");
  });

Answer 1

我会这样做的。

我们使用if (err) return callback(err); 来停止当前的异步函数并将错误发送到更高级别。

async.eachSeries(res, function (r, callback1) {

    let _id = r._id;
    let arr = r.campaignID;

    async.eachSeries(arr, function firstLevel (a, callback2) {

        let users = a.users;
        let id = a.id;
        let loop = Math.ceil(users / 1000);
        let limit = 0, offset = 0;

        // for loop is synchronous whereas mailjet is asynchronous -> usually bad idea to mix those two
        // instead try async.timesSeries()
        async.timesSeries(loop, function getSentMessages (n, callback3) {

            if (users > 1000) {
                limit = 1000;
                users -= limit;
            } else {
                limit = users;
            }
            console.log(n, limit, "limit", offset, "offset");

            var start = Date.now();
            while (Date.now() < start + 100) {} // this does nothing...

            // async.js doesn't flow well with Promises so request your resource with a callback function
            mailjet
                .get("messagesentstatistics")
                .request({ CampaignID: id, AllMessages: true, Limit: limit, Offset: offset })
                .request(function (err, result, body) {

                // stop everything if an error occurred; send the error back up
                if (err) return callback3(err);

                let data = result.body.Data;
                var loop = 0;

                async.eachSeries(data, secondLevel (jsonObj, callback4) {
                    let email = jsonObj.ToEmail;
                    jsonObj.retailer = r.retailer;
                    jsonObj.summary = 'f';
                    let tempObj = {};
                    tempObj[id] = jsonObj;
                    let options = { new: true };
                    let campId = id;
                    User.addCampaignResponse(email, campId, tempObj, options, function (err, results) {
                        // stop everything if an error occurred; send the error back up
                        if (err) return callback4(err);

                        console.log("added campaign response");

                        Campaign.updateResponse(_id, function (err, results2) {
                            // stop everything if an error occurred; send the error back up
                            if (err) return callback4(err);

                            console.log("updated campaign response");

                            callback4();
                        });
                    })
                }, callback3);

            }); // end of mailjet

            offset += limit;

        }, callback2); // end of async.timesSeries

    }, callback1); // end of async.eachOfSeries

}, function (err) {
    // if an error occurs anywhere, it should back here
    if (err) {
        console.log(err);
        return;
    }
    console.log("doneeeeee");
});

此外，使用有意义的变量和函数名称总是更好。

Answer 2

这是它的简化版

---

function secondLevel(value2, val, callback) {
    // ...
    User.addCampaignResponse(email, campId, tempObj, options, function (err, results) {
       if (err) {
          throw err;
       }

       Campaign.updateResponse(_id, function (err, results2) {
          // throw the error to the highest level if there is one
          if (err) throw err;

          // Finish the eachOfSeries if all is ok
          callback();
       });
    });
 }

---

function firstLevel(value1, i, callback) {
  // ...
  mailjet
    .get("messagesentstatistics")
    .request({
      // ...
    })
    .then((result) => {
      // ...
      async.eachOfSeries(data, secondLevel, function (err) {
        // throw the error to the highest level if there is one
        if (err) throw err;

        // Finish the eachOfSeries if all is ok
        callback();
      });
    })
    .catch((err) => {
      // throw the error to the highest level
      throw err;
    });
}

---

 //
 // Start of our program
 //
 async.eachOfSeries(res, function (value, camp, callback) {
     // ...
     async.eachOfSeries(arr, firstLevel, function (err) {
       // throw the error to the highest level if there is one
       if (!err) throw err;

       // Finish the eachOfSeries if all is ok
       callback();
     });
 },
 function (err) {
     if (err) {
        // Here we re done with an error
        // ...
        return;
     }

     // We did it, no, error
     // ...
 });