Node.js の非同期制御の使用例

今回は Node.js 非同期制御の使用例を紹介します。Node.js 非同期制御を使用する際の 注意事項 について、実際の事例を見てみましょう。

関数コールバックの使用に関する深い経験がない場合、これらの内容を読むのはまだ少し難しいです。 Node.js の独特な非同期特性により、「コールバック地獄」の問題が発生しました。この記事では、非同期フローの問題を解決する方法を詳しく記録しました。

記事が長くなりますが、これは非同期ストリーミング パターンの説明です。この記事では、単純な Web スパイダーの例を使用します。その機能は、指定された URL の Web ページ コンテンツをクロールし、それをプロジェクトに保存することです。記事全体のソース コード デモが表示されます。

1. ネイティブJavaScriptモード

この記事は初心者向けではないため、基本的な内容のほとんどは省略されます:

(spider_v1.js)

りー

上記のコードの流れはおおよそ次のようになります:

1. URLをファイル名に変換します

2. ファイル名が存在するかどうかを判断し、存在する場合は直接戻り、そうでない場合は次のステップに進みます

3. 本文を取得するリクエストを送信します

4. ファイルに本文を書き込みます

これはスパイダーの非常に単純なバージョンであり、URL のコンテンツをクロールすることしかできません。上記のコールバックがいかに面倒であるかがわかります。次に、最適化を開始します。

まず、if else メソッドを最適化することができます。言うまでもなく、これは比較効果です。 コードがこのように記述されると、ネストの層が 1 つ少なくなりますが、経験豊富なプログラマーは、このように記述するとエラーが強調されすぎると考えるでしょう。プログラミングの焦点は正しいデータを処理することにあるべきであり、これは にも存在します。可読性の要件。

もう 1 つの最適化は、上記のコードのスパイダー関数で、ダウンロードされたファイルと保存されたファイルを分割することができます。

(spider_v2.js)

りー

上記のコードは基本的にネイティブ最適化の結果ですが、このスパイダーの機能は単純すぎるため、特定の Web ページ内のすべての URL をクロールする必要があるため、シリアルおよびパラレルの問題が発生します。

(spider_v3.js)

りー

上記のコードには、前のコードよりも 2 つのコア関数が増えています。まず、補助クラス

const request = require("request");
const fs = require("fs");
const mkdirp = require("mkdirp");
const path = require("path");
const utilities = require("./utilities");
function spider(url, callback) {
  const filename = utilities.urlToFilename(url);
  console.log(`filename: ${filename}`);
  fs.exists(filename, exists => {
    if (!exists) {
      console.log(`Downloading ${url}`);
      request(url, (err, response, body) => {
        if (err) {
          callback(err);
        } else {
          mkdirp(path.dirname(filename), err => {
            if (err) {
              callback(err);
            } else {
              fs.writeFile(filename, body, err => {
                if (err) {
                  callback(err);
                } else {
                  callback(null, filename, true);
                }
              });
            }
          });
        }
      });
    } else {
      callback(null, filename, false);
    }
  });
}
spider(process.argv[2], (err, filename, downloaded) => {
  if (err) {
    console.log(err);
  } else if (downloaded) {
    console.log(`Completed the download of ${filename}`);
  } else {
    console.log(`${filename} was already downloaded`);
  }
});

を通じて特定の本体内のリンクを取得します。 内部実装については説明しません。他のコア コードは

/// before
if (err) {
  callback(err);
} else {
  callback(null, filename, true);
}
/// after
if (err) {
  return callback(err);
}
callback(null, filename, true);

です。 上記の小さなコードは、非同期シリアル化を実装するためのネイティブ パターンであると言えます。これらに加えて、この属性を通じてネストの概念も導入され、クロール レベルを制御できます。

この時点でシリアル機能の実装は完了しました。パフォーマンスを考慮して、並列クローリング機能を開発する必要があります。

(spider_v4.js)

りー

このコードも非常に単純で、2 つの主要な内容があります。 1 つは同時実行性を実現する方法です:

const request = require("request");
const fs = require("fs");
const mkdirp = require("mkdirp");
const path = require("path");
const utilities = require("./utilities");
function saveFile(filename, contents, callback) {
  mkdirp(path.dirname(filename), err => {
    if (err) {
      return callback(err);
    }
    fs.writeFile(filename, contents, callback);
  });
}
function download(url, filename, callback) {
  console.log(`Downloading ${url}`);
  request(url, (err, response, body) => {
    if (err) {
      return callback(err);
    }
    saveFile(filename, body, err => {
      if (err) {
        return callback(err);
      }
      console.log(`Downloaded and saved: ${url}`);
      callback(null, body);
    });
  })
}
function spider(url, callback) {
  const filename = utilities.urlToFilename(url);
  console.log(`filename: ${filename}`);
  fs.exists(filename, exists => {
    if (exists) {
      return callback(null, filename, false);
    }
    download(url, filename, err => {
      if (err) {
        return callback(err);
      }
      callback(null, filename, true);
    })
  });
}
spider(process.argv[2], (err, filename, downloaded) => {
  if (err) {
    console.log(err);
  } else if (downloaded) {
    console.log(`Completed the download of ${filename}`);
  } else {
    console.log(`${filename} was already downloaded`);
  }
});

上記のコードは、同時実行を実現するためのパターンと言えます。これはループトラバーサルを使用して実現されます。もう 1 つの重要な点は、同時実行であるため、fs.exists を使用すると問題が発生し、同じファイルが繰り返しダウンロードされる可能性があることです。その解決策は次のとおりです。 Map を使用して特定の URL をキャッシュします。その URL をキーとして使用する必要があります

同時実行の最大数を制限するという新しい要件があるため、ここでは最も重要だと思われる概念であるキューを紹介します。

• (task-Queue.js)

  const request = require("request");
  const fs = require("fs");
  const mkdirp = require("mkdirp");
  const path = require("path");
  const utilities = require("./utilities");
  function saveFile(filename, contents, callback) {
    mkdirp(path.dirname(filename), err => {
      if (err) {
        return callback(err);
      }
      fs.writeFile(filename, contents, callback);
    });
  }
  function download(url, filename, callback) {
    console.log(`Downloading ${url}`);
    request(url, (err, response, body) => {
      if (err) {
        return callback(err);
      }
      saveFile(filename, body, err => {
        if (err) {
          return callback(err);
        }
        console.log(`Downloaded and saved: ${url}`);
        callback(null, body);
      });
    })
  }
  /// 最大的启发是实现了如何异步循环遍历数组
  function spiderLinks(currentUrl, body, nesting, callback) {
    if (nesting === 0) {
      return process.nextTick(callback);
    }
    const links = utilities.getPageLinks(currentUrl, body);
    function iterate(index) {
      if (index === links.length) {
        return callback();
      }
      spider(links[index], nesting - 1, err => {
        if (err) {
          return callback(err);
        }
        iterate((index + 1));
      })
    }
    iterate(0);
  }
  function spider(url, nesting, callback) {
    const filename = utilities.urlToFilename(url);
    fs.readFile(filename, "utf8", (err, body) => {
      if (err) {
        if (err.code !== 'ENOENT') {
          return callback(err);
        }
        return download(url, filename, (err, body) => {
          if (err) {
            return callback(err);
          }
          spiderLinks(url, body, nesting, callback);
        });
      }
      spiderLinks(url, body, nesting, callback);
    });
  }
  spider(process.argv[2], 2, (err, filename, downloaded) => {
    if (err) {
      console.log(err);
    } else if (downloaded) {
      console.log(`Completed the download of ${filename}`);
    } else {
      console.log(`${filename} was already downloaded`);
    }
  });

上記のコードはキューの実装コードであり、タスクがキューに追加されるとすぐに実行されることがわかります。すぐに呼び出されますが、これは next がすぐに呼び出されるという意味です。

(spider_v5.js)

りー

したがって、同時実行の数を制限するには、spiderLinks メソッドのキューにタスクのトラバーサルを入れるだけです。これは比較的簡単です。

これまでのところ、ネイティブ JavaScript を使用して、比較的完全な機能を備えた Web スパイダーを実装してきました。この関数は、シリアルおよび同時実行の両方が可能で、同時実行数も制御できます。

2. 非同期ライブラリを使用する

把不同的功能放到不同的函数中，会给我们带来巨大的好处，async库十分流行，它的性能也不错，它内部基于callback。

(spider_v6.js)

const request = require("request");
const fs = require("fs");
const mkdirp = require("mkdirp");
const path = require("path");
const utilities = require("./utilities");
const series = require("async/series");
const eachSeries = require("async/eachSeries");
function download(url, filename, callback) {
  console.log(`Downloading ${url}`);
  let body;
  series([
    callback => {
      request(url, (err, response, resBody) => {
        if (err) {
          return callback(err);
        }
        body = resBody;
        callback();
      });
    },
    mkdirp.bind(null, path.dirname(filename)),
    callback => {
      fs.writeFile(filename, body, callback);
    }
  ], err => {
    if (err) {
      return callback(err);
    }
    console.log(`Downloaded and saved: ${url}`);
    callback(null, body);
  });
}
/// 最大的启发是实现了如何异步循环遍历数组
function spiderLinks(currentUrl, body, nesting, callback) {
  if (nesting === 0) {
    return process.nextTick(callback);
  }
  const links = utilities.getPageLinks(currentUrl, body);
  if (links.length === 0) {
    return process.nextTick(callback);
  }
  eachSeries(links, (link, cb) => {
    "use strict";
    spider(link, nesting - 1, cb);
  }, callback);
}
const spidering = new Map();
function spider(url, nesting, callback) {
  if (spidering.has(url)) {
    return process.nextTick(callback);
  }
  spidering.set(url, true);
  const filename = utilities.urlToFilename(url);
  fs.readFile(filename, "utf8", (err, body) => {
    if (err) {
      if (err.code !== 'ENOENT') {
        return callback(err);
      }
      return download(url, filename, (err, body) => {
        if (err) {
          return callback(err);
        }
        spiderLinks(url, body, nesting, callback);
      });
    }
    spiderLinks(url, body, nesting, callback);
  });
}
spider(process.argv[2], 1, (err, filename, downloaded) => {
  if (err) {
    console.log(err);
  } else if (downloaded) {
    console.log(`Completed the download of ${filename}`);
  } else {
    console.log(`${filename} was already downloaded`);
  }
});

在上边的代码中，我们只使用了async的三个功能：

const series = require("async/series"); // 串行
const eachSeries = require("async/eachSeries"); // 并行
const queue = require("async/queue"); // 队列

由于比较简单，就不做解释了。async中的队列的代码在(spider_v7.js)中，和上边我们自定义的队列很相似，也不做更多解释了。

3.Promise

Promise是一个协议，有很多库实现了这个协议，我们用的是ES6的实现。简单来说promise就是一个约定，如果完成了，就调用它的resolve方法，失败了就调用它的reject方法。它内有实现了then方法，then返回promise本身，这样就形成了调用链。

其实Promise的内容有很多，在实际应用中是如何把普通的函数promise化。这方面的内容在这里也不讲了，我自己也不够格

(spider_v8.js)

const utilities = require("./utilities");
const request = utilities.promisify(require("request"));
const fs = require("fs");
const readFile = utilities.promisify(fs.readFile);
const writeFile = utilities.promisify(fs.writeFile);
const mkdirp = utilities.promisify(require("mkdirp"));
const path = require("path");
function saveFile(filename, contents, callback) {
  mkdirp(path.dirname(filename), err => {
    if (err) {
      return callback(err);
    }
    fs.writeFile(filename, contents, callback);
  });
}
function download(url, filename) {
  console.log(`Downloading ${url}`);
  let body;
  return request(url)
    .then(response => {
      "use strict";
      body = response.body;
      return mkdirp(path.dirname(filename));
    })
    .then(() => writeFile(filename, body))
    .then(() => {
      "use strict";
      console.log(`Downloaded adn saved: ${url}`);
      return body;
    });
}
/// promise编程的本质就是为了解决在函数中设置回调函数的问题
/// 通过中间层promise来实现异步函数同步化
function spiderLinks(currentUrl, body, nesting) {
  let promise = Promise.resolve();
  if (nesting === 0) {
    return promise;
  }
  const links = utilities.getPageLinks(currentUrl, body);
  links.forEach(link => {
    "use strict";
    promise = promise.then(() => spider(link, nesting - 1));
  });
  return promise;
}
function spider(url, nesting) {
  const filename = utilities.urlToFilename(url);
  return readFile(filename, "utf8")
    .then(
      body => spiderLinks(url, body, nesting),
      err => {
        "use strict";
        if (err.code !== 'ENOENT') {
          /// 抛出错误，这个方便与在整个异步链的最后通过呢catch来捕获这个链中的错误
          throw err;
        }
        return download(url, filename)
          .then(body => spiderLinks(url, body, nesting));
      }
    );
}
spider(process.argv[2], 1)
  .then(() => {
    "use strict";
    console.log('Download complete');
  })
  .catch(err => {
    "use strict";
    console.log(err);
  });

可以看到上边的代码中的函数都是没有callback的，只需要在最后catch就可以了。

在设计api的时候，应该支持两种方式，及支持callback，又支持promise

function asyncpision(pidend, pisor, cb) {
  return new Promise((resolve, reject) => {
    "use strict";
    process.nextTick(() => {
      const result = pidend / pisor;
      if (isNaN(result) || !Number.isFinite(result)) {
        const error = new Error("Invalid operands");
        if (cb) {
          cb(error);
        }
        return reject(error);
      }
      if (cb) {
        cb(null, result);
      }
      resolve(result);
    });
  });
}
asyncpision(10, 2, (err, result) => {
  "use strict";
  if (err) {
    return console.log(err);
  }
  console.log(result);
});
asyncpision(22, 11)
  .then((result) => console.log(result))
  .catch((err) => console.log(err));

4.Generator

Generator很有意思，他可以让暂停函数和恢复函数，利用thunkify和co这两个库，我们下边的代码实现起来非常酷。

(spider_v9.js)

const thunkify = require("thunkify");
const co = require("co");
const path = require("path");
const utilities = require("./utilities");
const request = thunkify(require("request"));
const fs = require("fs");
const mkdirp = thunkify(require("mkdirp"));
const readFile = thunkify(fs.readFile);
const writeFile = thunkify(fs.writeFile);
const nextTick = thunkify(process.nextTick);
function* download(url, filename) {
  console.log(`Downloading ${url}`);
  const response = yield request(url);
  console.log(response);
  const body = response[1];
  yield mkdirp(path.dirname(filename));
  yield writeFile(filename, body);
  console.log(`Downloaded and saved ${url}`);
  return body;
}
function* spider(url, nesting) {
  const filename = utilities.urlToFilename(url);
  let body;
  try {
    body = yield readFile(filename, "utf8");
  } catch (err) {
    if (err.code !== 'ENOENT') {
      throw err;
    }
    body = yield download(url, filename);
  }
  yield spiderLinks(url, body, nesting);
}
function* spiderLinks(currentUrl, body, nesting) {
  if (nesting === 0) {
    return nextTick();
  }
  const links = utilities.getPageLinks(currentUrl, body);
  for (let i = 0; i < links.length; i++) {
    yield spider(links[i], nesting - 1);
  }
}
/// 通过co就自动处理了回调函数，直接返回了回调函数中的参数，把这些参数放到一个数组中，但是去掉了err信息
co(function* () {
  try {
    yield spider(process.argv[2], 1);
    console.log('Download complete');
  } catch (err) {
    console.log(err);
  }
});

相信看了本文案例你已经掌握了方法，更多精彩请关注php中文网其它相关文章！

推荐阅读：

以上がNode.js の非同期制御の使用例の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。