javascript - node.js async.each 回调，我怎么知道它什么时候完成？

Question

我正在努力研究 Node.js 和一些异步操作。在下面的代码中，我获取一些 RSS 提要并存储找到的文章(如果我以前没有存储过它们的话)。该代码可以工作并存储新文章。但是，我不确定如何更改它以便我知道何时完成对所有文章的解析。例如，每次达到限制时(在 10 篇文章或 5 个提要之后)都会调用每个 async.eachLimit 的回调。那么我怎么知道它们何时完成？

var FeedParser = require('feedparser');
var request = require('request');
var mysql = require('mysql');
var async = require('async');

var connection = mysql.createConnection({
        host :  'localhost',
        user : 'someuser',
        password : 'somepass',
        database : 'somedb'
});

connection.connect();

connection.query('SELECT * FROM rssfeed', function(err, rows, fields) {
        if(err == null){
                async.eachLimit(rows, 5, parseFeed, function(err) {
                        if(! err) {
                                //horray
                        } else {
                                console.log(err);
                        }
                });
        }
});

function parseFeed(feed, callback) {
        var articles = [];
        request(feed.link)
        .pipe(new FeedParser())
        .on('error', function(error) {
                callback(error);
        })
        .on('meta', function(meta) {
        })
        .on('readable', function() {
                var stream = this, item;
                item = stream.read();
                if(item != null) {
                        item.rssfeed_id = feed.id;
                        articles.push(item);
                }
        })
        .on('finish', function() {
                async.eachLimit(articles, 10, parseArticle, function(err) {
                        if(! err) {
                                console.log('article each callback');
                        } else {
                                callback(error);
                        }
                });
        });
        callback();
}
function parseArticle(item, callback) {
        if(item.hasOwnProperty('rssfeed_id') && item.hasOwnProperty('guid') && item.hasOwnProperty('link') && item.hasOwnProperty('title')){
                connection.query('SELECT * FROM rssarticle WHERE rssfeed_id = ? AND guid = ?', [item.rssfeed_id, item.guid], function(err, rows, fields) {
                        if(rows.length == 0){
                                connection.query('INSERT INTO rssarticle SET ?', {
                                        rssfeed_id: item.rssfeed_id,
                                        link: item.link,
                                        title: item.title,
                                        description: item.description,
                                        publish_date: item.pubDate,
                                        guid: item.guid
                                }, function(err, result){
                                        if(err != null){
                                                console.log(err);
                                        }
                                });
                        }
                });
        }
        callback();
}

Answer 1

首先，您过早地抢先调用回调。

function parseFeed(feed, callback) {
  request
    .streamStuff()
    .streamStuff()
    .streamStuff();

  callback();
}

您不应该在完成之前调用 callback。否则，您的“完成”方法将被调用，但您的异步代码实际上仍在运行。

所以不要这样做:

.on('finish', function() {
    async.eachLimit(articles, 10, parseArticle, function(err) {
        if(! err) {
            console.log('article each callback');
        } else {
            callback(error);
        }
    });
});
callback();

就这样

.on('finish', function() {
    async.eachLimit(articles, 10, parseArticle, function(err) {
        if(! err) {
            // assuming this is a stub and really ends up doing `callback();`
            console.log('article each callback');
        } else {
            callback(error);
        }
    });
});