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

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

配信処理でのStream APIの活用事例

　さて、ここからはPusna-RSの配信機能でのStreamの使い方について紹介します。以下が配信機能の構成図です。

　配信機能では「executorStream」というStreamの中で処理を行っています。executorStreamの中で以下の5つの機能のStreamをpipeで連結させ、一連の配信の流れを実現します。

• reader
  全件・ID指定・検索の配信タイプに応じて、DynamoDBまたはelasticsearchからデータを抽出する
• transfer
  配信タイプに応じてreaderから受け取ったデータを変換する
• notification
  transferでまとめたデータをiOSの場合APNs、Androidの場合GCMに送信する
• result
  notificationから受け取った配信結果の件数をカウントし、最終的な配信件数をDynamoDBに保存する
• registration
  notificationから受け取った配信結果を順次ビッグデータ基盤に格納するためのSNSに送信する

　前述の通り、各機能では配信パターンに応じて複数の実装を切り替えています。readerでは全件の場合scanStream、ID指定の場合queryStream、検索の場合searchStreamといった形です。

　また、notificationからresultとregistrationの2つに分岐しています。どちらも配信結果を記録する機能ですが、resultは要求に対する最終的な結果を格納するために使っており、registrationは配信一つ一つの結果をビッグデータ基盤に送るためのものであり、用途が異なるためnotificationから分岐をさせています。

　以下がexecutorStreamの流れを簡略化したサンプルです。

var readerStream = new SearchStream(query);            // elasticsearchからデータ抽出を行うStream
var transferStream = new PushTransferStream();         // 受け取ったそばから送信するStream
var notificationStream = new ApnsNotificationStream(); // APNsにデータ送信を行うStream
var resultStream = new ResultStream();                 // DynamoDBに結果を保存するStream
var registrationStream = new RegistrationStream();     // ビッグデータ基盤に結果を送るStream
 
// 各Streamを連結させる 
readerStream.pipe(transferStream)
            .pipe(notificationStream)
            .pipe(resultStream);
 
// notificationにregistrationを別に連結させる
notificationStream.pipe(registrationStream);

Stream APIのpauseとresumeで全体の速度調整

　上記のStreamにおいて、全ての処理速度が同じ場合は何の問題もないのですが、実際は各Streamにより処理速度は異なります。例えば、DynamoDBからの抽出は非常に高速ですが、GCMへの送信は一回一回HTTP通信が必要となるため、速度は大きく落ちます。

　Stream APIには、この速度の違いを吸収するために「pause」「resume」という機能があります（Stream2の場合は仕様が異なりますが動きは同様です）。

　以下はreaderから抽出したデータを2回変換をしながらnotificationに送る処理での例です。

　notificationstream以外は100のスピードで処理ができ、notificationはその半分しかスピードが出ないとすると、notificationで処理しきれなかった50がバッファーにたまっていきます。

　処理スピードはずっと変わらないため、バッファーはどんどんたまっていってしまい、たまり過ぎるとメモリがあふれるまで使ってしまいます。

　このままでは溢れてしまうのでnotificationStreamはpauseを実行して読み込みを停止します。

　pauseはpipeしている手前のStreamに伝播し、読み込み全体を一時停止させることができます。

　これによりnotificationStreamはたまったバッファーの処理を先に行えます。

　そしてバッファーの処理が完了したタイミングでresumeを実行することで、読み込みを再開できます。

　実際のシステムでは速度はもっとバラバラになりますが、この仕組みによりシステム全体の速度をボトルネックとなる箇所の速度と同じにすることができます。

　Pusna-RSではAPNs／GCMへの送信箇所に最も時間がかかるため、notificationの性能が配信速度となります。notificationが高速化することで送信時間全体が短縮されるため、個別のチューニングも行っています。

• APNs
  • アプリごとにコネクションプールを作り、一度の送信に複数のコネクションを使って送信
  • 一定件数単位にまとめて送信
• GCM
  • HTTP通信が必要なためKeepAliveで接続の回数を減らし、一度の送信に複数のコネクションを使って送信
  • 上限となる1000件単位にまとめて送信

大量のプッシュ通知を送ると起こる問題

　今回はPusna-RSにおける配信機能についてStream APIの活用を中心に紹介しました。

　これらの作り込みによりプッシュ通知を秒間1万4000ほど送ることができています。しかし、実は運用中に速くし過ぎたことによる障害も発生しました。プッシュ通知は開封率が高いため、一度に大量のプッシュ通知を行った際に大量の開封によりアプリケーションのサーバーサイドが高負荷状態になってしまう事態が起こりました。これにより、急きょプッシュの送信速度をコントロールする機能をリリースしています。

　次回は、このようにいろいろの問題が起こる運用をどのように行っているかについて紹介します。

「大規模プッシュ通知基盤大解剖」バックナンバー

• 24時間途切れないサービスで有効なImmutable Infrastructureの運用方法
• Node.jsのStream APIで大量プッシュ通知を高速化するテクニック
• 大量データ処理時に知っておきたいAmazon DynamoDB活用テクニック4選
• プッシュ通知の基礎知識＆秒間1万を超えるプッシュ通知基盤のアーキテクチャと仕組みとは