Node.jsのStream APIで大量プッシュ通知を高速化するテクニック：大規模プッシュ通知基盤大解剖（3）（1/2 ページ）

大規模プッシュ通知基盤について、「Pusna-RS」の実装事例を基にアーキテクチャや運用を解説する連載。今回は、Node.jsのStream APIや非同期処理の基本をあらためて解説し、pauseとresumeで全体の速度調整を行う方法などを紹介します。

[宮川典久，リクルートテクノロジーズ]

連載目次

プッシュ通知を秒間1万4000ほど送るには

　前回の「大量データ処理時に知っておきたいAmazonDyamoDB活用テクニック4選」では、Pusna−RSのデータ永続化に使っているDynamoDBの活用テクニックについて解説しました。今回は、そのDynamoDBからAPNs／GCMへのデータ送信までを高速化させた方法を紹介します。

Pusna-RSの配信機能とは

　連載第1回「プッシュ通知の基礎知識＆秒間1万を超えるプッシュ通知基盤のアーキテクチャと仕組みとは」の際に概要をお伝えしましたが、配信機能はPusna-RSの中で実際にプッシュ通知を行うための機能で、以下の【3】に当たる箇所です。

　前回全件抽出のためにDynamoDBの並列スキャンを活用していることを紹介しましたが、プッシュを高速化するためにはデータ抽出だけが速くてもダメで、APNs／GCMまで送る処理全体が最適化されている必要があります。

　Pusna-RSではこの一連の流れを最適化させるためにNode.jsのStream APIを活用しています。今回は、このStream APIの活用について紹介していきます。

あらためてNode.jsのStream APIとは

　Stream APIはNode.jsでデータの流れを扱うためのAPIです。Stream API自体の説明の前に、まずは前提としてNode.jsの非同期処理について説明します。

非同期処理とは

　Node.jsにおける非同期処理は、時間のかかる処理にコールバック関数を登録し、処理が完了した時点で渡したコールバック関数を実行する仕組みです。Node.jsはシングルスレッドで動作しているため、多くの処理を非同期で実行する必要があります。

　特に、I/O処理では処理中に終わりを待たずに別の処理を行えるため、アプリケーションの速度向上に非常に有効です。以下ファイル読み込みの簡単なサンプルで同期処理と非同期処理の違いについて説明します。

require('fs');
 
console.log(‘start’);
var data1  =  fs.readFileSync(‘hoge.txt’);
console.log(‘file1:’ + data1);
var data2  =  fs.readFileSync(‘fuga.txt’);
console.log(‘file2:’ + data2);
console.log(‘end’);

同期処理

start
file1: hoge
file2: fuge
end

結果

　fs.readFileSyncはファイル読み込みを同期で行うAPIです。上記の場合、ファイル読み込みが完了するまで待ってから先に進むため、コードに記載している通り、上から順番にconsole.logが出力されます。

require('fs');
 
console.log(‘start’);
fs.readFile(‘hoge.txt’, function(err, data) {
  console.log(‘file1:’ + data);
});
fs.readFile(‘fuga.txt’, function(err, data) {
  console.log(‘file2:’ + data);
});
console.log(‘end’);

非同期処理

start
end
file1: hoge
file2: fuge

結果

　一方fs.readFileはファイル読み込みを非同期で行うAPIです。fs.readFileSyncと異なり処理を開始した段階で次のステップへ処理が進みます。そのため、結果上も先にconsole.logの「end」が出力されます。読み込み処理が完了したタイミングで第二引数に渡しているコールバック関数が実行され、読み込んだデータに対する処理が実行されます。

　同期処理と非同期処理によるI/O処理の違いを図にすると以下のようになります。

　非同期処理を使うことでI/O処理の多重化やI/O処理中に別の処理を行うなどのことが容易に行えるため、性能につながりやすいです。一方処理が上から流れるわけではなくなるため、実装に当たっては慣れが必要な面もあります。

　さて、サンプルのようにファイル読み込みが全て終わってから処理を行えばいいものは非同期APIだけで問題ないのですが、Pusna-RSのように「DynamoDBから抽出をしながら送信を行いたい」という用途は非同期APIだけでは実現できません。

　このように「データを抽出してから変換し、出力する」という一連の処理を効率よく実現するための仕組みがStream APIです。

Stream APIとは

　Stream APIはデータの流れを表すAPIです。以下はファイルを1バイトずつ読み込むサンプルです。

var readableStream = fs.createReadStream(‘hoge.txt‘,
  {bufferSize: 1}); 
 
readableStream.on('data', 
  function(data) {
   console.log(data); 
});
 
readableStream.on('end', 
  function() {
   console.log('end'); 
});

　先ほどのreadFileとは異なり、1バイトずつ読み込みを行いながら操作を行えます。

　さらに、Streamは他のStreamと連結できます。Streamはpipeメソッドを持っており、Linuxのpipeと同じように自分が処理をした結果を順次、次のStreamに渡すことになります。

　Stream APIは大きくReadableとWritableの2種類があります。Readableは上記サンプルのようにデータを読み込むStreamで、Writableはデータを書き込むStreamです。

　応用系としてReadableとWritableを合わせて、読み込み書き込み両方を行うStreamを作ることも可能です。これらの組み合わせによりデータの抽出、変換、送信という一連の流れを実現できます。

Stream APIのバージョンStream1、Stream2、Stream3の違い

　Stream APIは現在Stream1、Stream2、Stream3という複数のバージョンがあり、それぞれで仕様が異なりますが基本的な流れは同様です。

　Pusna-RSではNode.js v0.8系を選択したため、Stream1を使用しています。

　Node.js v0.10系の場合Stream2となりStream1は使用できません。

　Stream3はNode.jsの次のバージョンのv0.12またはNode.jsからフォークされた「io.js」で使えます。Stream3では、Stream1とStream2が統合され、両方の機能を使うことができます。