javascript - forEach 完成后运行回调函数

Question

在项目中，我有一个遍历 url 列表的循环。它从每个 url 下载文件并对下载的文件进行一些后期处理。

完成所有过程(下载过程和后期过程)后，我想执行一个回调函数。因为后处理包括一些流任务，所以它有关闭事件。如果可以识别最后一项，我可以将回调函数传递给关闭事件。但是，由于循环是异步的，我无法跟踪最后完成了哪个项目。

现在，我使用 5 秒超时来确保回调在整个过程之后执行。显然，这是不可持续的。处理此问题的好方法是什么？

循环代码:

exports.processArray = (items, process, callback) => {
    var todo = items.concat();
    setTimeout(function() {
        process(todo.shift());
        if(todo.length > 0) {
          // execute download and post process each second
          // however it doesn't guarantee one start after previous one done
          setTimeout(arguments.callee, 1000);
        } else {
          setTimeout(() => {callback();}, 5000);
        }
    }, 1000);
};

processArray(
  // First param, the array
  urlList,
  // Second param, download and post process
  (url) => {
    if(url.startsWith('http')) {
      getDataReg(url, uid);
    }
    else if(url.startsWith('ftp')) {
      getDataFtp(url, uid);
    }
    else {
      console.log('not a valid resource');
    }
  },
  // Third param, callback to be executed after all done
  () => {
    Request.get(`${config.demouri}bound=${request.query.boundary};uid=${uid}`, {
      method: 'GET',
      auth: auth
    })
    .on('response', (response) => {
      console.log('response event emmits');
      zipFiles(uid)
      .then((path) => {
        reply.file(path, { confine: false, filename: uid + '.zip', mode: 'inline'}).header('Content-Disposition');
      });
    });
  }
);

下载和发布过程:

exports.getDataFtp = (url, uid) => {
  console.log('get into ftp');
  var usefulUrl = url.split('//')[1];
  var spliter = usefulUrl.indexOf('/');
  var host = usefulUrl.substring(0, spliter);
  var dir = usefulUrl.substring(spliter+1, usefulUrl.length);
  var client = new ftp();
  var connection = {
    host: host
  };
  var fileNameStart = dir.lastIndexOf('/') + 1;
  var fileNameEnd = dir.length;
  var fileName = dir.substring(fileNameStart, fileNameEnd);
  console.log('filename: ', fileName);

  client.on('ready', () => {
    console.log('get into ftp ready');
    client.get(dir, (err, stream) => {
      if (err) {
        console.log('get file err:', err);
        return;
      } else{
        console.log('get into ftp get');
        stream.pipe(fs.createWriteStream(datadir + `download/${uid}/${fileName}`));
        stream.on('end', () => {
          console.log('get into ftp close');
          unzipData(datadir + `download/${uid}/`, fileName, uid);
          client.end();
        });
      }
    });
  });
  client.connect(connection);
};

exports.getDataReg = (url, uid) => {
  console.log('get into http');
    var fileNameStart = url.lastIndexOf('/') + 1;
  var fileNameEnd = url.length;
  var fileName = url.substring(fileNameStart, fileNameEnd);
    var file = fs.createWriteStream(datadir + `download/${uid}/${fileName}`);
    if (url.startsWith('https')) {
    https.get(url, (response) => {
      console.log('start piping file');
      response.pipe(file);
      file.on('finish', () => {
        console.log('get into http finish');
        unzipData(datadir + `download/${uid}/`, fileName, uid);
      });
    }).on('error', (err) => { // Handle errors
      fs.unlink(datadir + `download/${uid}/${fileName}`);
      console.log('download file err: ', err);
    });
    } else {
    http.get(url, (response) => {
      console.log('start piping file');
      response.pipe(file);
      file.on('finish', () => {
        unzipData(datadir + `download/${uid}/`, fileName, uid);
      });
    }).on('error', (err) => {
      fs.unlink(datadir + `download/${uid}/${fileName}`);
      console.log('download file err: ', err);
    });
    }
};

function unzipData(path, fileName, uid) {
  console.log('get into unzip');
  console.log('creating: ', path + fileName);
    fs.createReadStream(path + fileName)
    .pipe(unzip.Extract({path: path}))
    .on('close', () => {
    console.log('get into unzip close');
    var filelist = listFile(path);
    filelist.forEach((filePath) => {
      if (!filePath.endsWith('.zip')) {
        var components = filePath.split('/');
        var component = components[components.length-1];
        mv(filePath, datadir + `processing/${uid}/${component}`, (err) => {
          if(err) {
            console.log('move file err: ');
          } else {
            console.log('move file done');
          }
        });
      }
    });
    fs.unlink(path + fileName, (err) => {});
    });
}

Answer 1

After the all the process done (both download process and post process), I want to execute a callback function.

关于一系列异步进程的有趣之处在于，您永远无法知道所有进程何时准确完成。因此，为回调设置超时是一种快速而肮脏的方法，但它确实不可靠。

您可以改用计数器 来解决这个问题。 假设您要执行 10 个操作。开始时，您将计数器设置为 10 counter = 10 在每个过程完成后，无论如何(成功或失败)，您都可以将计数器减 1，如 counter - = 1 紧接着您可以检查计数器是否为 0，如果是，则意味着所有过程都已完成，我们到达了终点。您现在可以安全地运行回调函数，例如 if(counter === 0) callback();

---

如果我是你，我会这样做:

*请注意，被调用的进程应该返回一个 promise ，这样我就可以知道它何时完成(再次不管如何完成)

*如果您需要有关 promise 的帮助，这篇有用的文章可能会对您有所帮助:https://howtonode.org/promises

*还有一件事，您应该避免使用 arguments.callee，因为它已被弃用。这就是为什么 Why was the arguments.callee.caller property deprecated in JavaScript?

exports.processArray = (items, process, callback) => {
    var todo = [].concat(items);
    var counter = todo.length;

    runProcess();

    function runProcess() {
      // Check if the counter already reached 0
      if(checkCounter() === false) {
        // Nope. Counter is still > 0, which means we got work to do.
        var processPromise = process(todo.shift());

        processPromise
          .then(function() {
            // success
          })
          .catch(function() {
            // failure
          })
          .finally(function() {
            // The previous process is done. 
            // Now we can go with the next one.
            --counter;
            runProcess();
          })
      }
    };

    function checkCounter() {
      if(counter === 0) {
        callback();
        return true;
      } else {
        return false;
      }
    }
};