在node.js中使用Async瀑布

Question

我有两个函数,我正在异步运行.我想用瀑布模型来编写它们.问题是,我不知道怎么做..

这是我的代码:

var fs = require('fs');
function updateJson(ticker, value) {
  //var stocksJson = JSON.parse(fs.readFileSync("stocktest.json"));
  fs.readFile('stocktest.json', function(error, file) {
    var stocksJson =  JSON.parse(file);

    if (stocksJson[ticker]!=null) {
      console.log(ticker+" price : " + stocksJson[ticker].price);
      console.log("changing the value...")
      stocksJson[ticker].price =  value;

      console.log("Price after the change has been made -- " + stocksJson[ticker].price);
      console.log("printing the the Json.stringify")
      console.log(JSON.stringify(stocksJson, null, 4));
      fs.writeFile('stocktest.json', JSON.stringify(stocksJson, null, 4), function(err) {  
        if(!err) {
          console.log("File successfully written");
        }
        if (err) {
          console.error(err);
        }
      }); //end of writeFile
    } else {
      console.log(ticker + " doesn't exist on the json");
    }
  });
} // end of updateJson 

任何想法我怎么能用瀑布写它,所以我能够控制它？请给我写一些例子,因为我是node.js的新手

Answer 1

首先确定步骤并将它们写为异步函数(采用回调参数)

• 读取文件

  function readFile(readFileCallback) {
      fs.readFile('stocktest.json', function (error, file) {
          if (error) {
              readFileCallback(error);
          } else {
              readFileCallback(null, file);
          }
      });
  }
  

• 处理文件(我删除了示例中的大部分console.log)

  function processFile(file, processFileCallback) {
      var stocksJson = JSON.parse(file);
      if (stocksJson[ticker] != null) {
          stocksJson[ticker].price = value;
          fs.writeFile('stocktest.json', JSON.stringify(stocksJson, null, 4), function (error) {
              if (err) {
                  processFileCallback(error);
              } else {
                  console.log("File successfully written");
                  processFileCallback(null);
              }
          });
      }
      else {
          console.log(ticker + " doesn't exist on the json");
          processFileCallback(null); //callback should always be called once (and only one time)
      }
  }
  

请注意,我没有在这里进行特定的错误处理,我将利用async.waterfall来集中在同一个地方的错误处理.

另外要注意,如果你在异步函数中有(if/else/switch/...)分支,它总是调用回调一次(并且只有一次).

使用async.waterfall插入所有内容

async.waterfall([
    readFile,
    processFile
], function (error) {
    if (error) {
        //handle readFile error or processFile error here
    }
});

干净的例子

之前的代码过于冗长,使解释更加清晰.这是一个完整的清理示例:

async.waterfall([
    function readFile(readFileCallback) {
        fs.readFile('stocktest.json', readFileCallback);
    },
    function processFile(file, processFileCallback) {
        var stocksJson = JSON.parse(file);
        if (stocksJson[ticker] != null) {
            stocksJson[ticker].price = value;
            fs.writeFile('stocktest.json', JSON.stringify(stocksJson, null, 4), function (error) {
                if (!err) {
                    console.log("File successfully written");
                }
                processFileCallback(err);
            });
        }
        else {
            console.log(ticker + " doesn't exist on the json");
            processFileCallback(null);
        }
    }
], function (error) {
    if (error) {
        //handle readFile error or processFile error here
    }
});

我留下了函数名称,因为它有助于提高可读性,并有助于使用chrome调试器等工具进行调试.

如果你使用下划线(在npm上),你也可以用第一个函数替换_.partial(fs.readFile, 'stocktest.json')

Answer 2

首先,请务必阅读有关的文档async.waterfall.

现在,关于瀑布控制流程有几个关键部分:

1. 控制流由作为第一个参数的调用函数数组指定,当流完成作为第二个参数时指定"完整"回调.
2. 函数数组是串行调用的(而不是并行调用).
3. 如果err在流数组中的任何操作中遇到错误(通常是命名的),它将短路并立即调用"完成"/"完成"/"完成" callback.
4. 来自先前执行的函数的参数按顺序应用于控制流中的下一个函数,并且提供"中间"回调作为最后一个参数.注意:第一个函数只有这个"中间"回调,"完整"回调将包含控制流中最后一个被调用函数的参数(考虑到任何错误),但是err前面加一个参数而不是"中间"附加的回调.
5. cbAsync当你准备继续前进时,应该调用每个单独操作的回调(我在我的例子中称之为):第一个参数是错误,如果有的话,第二个参数(第三个,第四个......等等)参数将是您要传递给后续操作的任何数据.

第一个目标是让您的代码几乎逐字逐句地工作async.waterfall.我决定删除所有console.log语句并简化错误处理.这是第一次迭代(未经测试的代码):

var fs = require('fs'),
    async = require('async');

function updateJson(ticker,value) {
    async.waterfall([ // the series operation list of `async.waterfall`
        // waterfall operation 1, invoke cbAsync when done
        function getTicker(cbAsync) {
            fs.readFile('stocktest.json',function(err,file) {
                if ( err ) {
                    // if there was an error, let async know and bail
                    cbAsync(err);
                    return; // bail
                }
                var stocksJson = JSON.parse(file);
                if ( stocksJson[ticker] === null ) {
                    // if we don't have the ticker, let "complete" know and bail
                    cbAsync(new Error('Missing ticker property in JSON.'));
                    return; // bail
                }
                stocksJson[ticker] = value;
                // err = null (no error), jsonString = JSON.stringify(...)
                cbAsync(null,JSON.stringify(stocksJson,null,4));    
            });
        },
        function writeTicker(jsonString,cbAsync) {
            fs.writeFile('stocktest.json',jsonString,function(err) {
                cbAsync(err); // err will be null if the operation was successful
            });
        }
    ],function asyncComplete(err) { // the "complete" callback of `async.waterfall`
        if ( err ) { // there was an error with either `getTicker` or `writeTicker`
            console.warn('Error updating stock ticker JSON.',err);
        } else {
            console.info('Successfully completed operation.');
        }
    });
}

第二次迭代将操作流程再划分.它将它放入较小的单操作导向代码块中.我不打算发表评论,它不言而喻(再次,未经测试):

var fs = require('fs'),
    async = require('async');

function updateJson(ticker,value,callback) { // introduced a main callback
    var stockTestFile = 'stocktest.json';
    async.waterfall([
        function getTicker(cbAsync) {
            fs.readFile(stockTestFile,function(err,file) {
                cbAsync(err,file);
            });
        },
        function parseAndPrepareStockTicker(file,cbAsync) {
            var stocksJson = JSON.parse(file);
            if ( stocksJson[ticker] === null ) {
                cbAsync(new Error('Missing ticker property in JSON.'));
                return;
            }
            stocksJson[ticker] = value;
            cbAsync(null,JSON.stringify(stocksJson,null,4));
        },
        function writeTicker(jsonString,cbAsync) {
            fs.writeFile('stocktest.json',jsonString,,function(err) {
                cbAsync(err);
            });
        }
    ],function asyncComplete(err) {
        if ( err ) {
            console.warn('Error updating stock ticker JSON.',err);
        }
        callback(err);
    });
}

最后一次迭代通过使用一些bind技巧来减少调用堆栈并增加可读性(IMO),这也是未经测试的:

var fs = require('fs'),
    async = require('async');

function updateJson(ticker,value,callback) {
    var stockTestFile = 'stocktest.json';
    async.waterfall([
        fs.readFile.bind(fs,stockTestFile),
        function parseStockTicker(file,cbAsync) {
            var stocksJson = JSON.parse(file);
            if ( stocksJson[ticker] === null ) {
                cbAsync(new Error('Missing ticker property in JSON.'));
                return;
            }
            cbAsync(null,stocksJson);
        },
        function prepareStockTicker(stocksJson,cbAsync) {
            stocksJson[ticker] = value;
            cbAsync(null,JSON.stringify(stocksJson,null,4));
        },
        fs.writeFile.bind(fs,stockTestFile)
    ],function asyncComplete(err) {
        if ( err ) {
            console.warn('Error updating stock ticker JSON.',err);
        }
        callback(err);
    });
}