予定時刻になったら通知する。

今でこそあたりまえの機能ですが、Aipoがこの機能をリリースしたのは2016年でした。（当時のアップデート情報）

実行速度やスケーラビリティを考慮したしくみになっており、Aipoのリアルタイム通信や大規模配信の基礎となっていますのでご紹介します。

解決したい問題

• 会議の時間になったら教えてほしい
• 大事な作業の開始時刻を忘れないようにしたい

このような問題を解決するため予定時刻（あるいは予定の5分前、15分前など）になったら通知をしてほしい、という要望を多くのユーザーからいただき開発に取りかかりました。

その中で、次のような技術的課題があることがわかりました。

技術的課題

分散したデータベース

Aipoはマルチテナント型のSaaSサービスです。マルチテナント型にもさまざまな種類がありますが、Aipoはブリッジモデルを採用しています。

参照：
AWS 上のマルチテナント SaaS 環境におけるセキュリティプラクティス
SaaS Tenant Isolation Strategies

Aipoではアプリケーションレイヤーはすべてのユーザーで共通ですが、データベースはワークスペース（契約企業）ごと別にすることで、別のワークスペースのデータを参照、あるいは更新が起きないようにしています。

すべてのワークスペースでデータベースが1つであれば、1回のSQLでリマインダーを送信する対象の予定を取得できます。しかしワークスペースごとにデータベースがあるため、データベースの数だけSQLを実行して対象の予定を取得する必要があります。

Aipoの予定は5分きざみで登録できるため、この方法だと5分ごとにすべてのデータベースをチェックすることになります。しかしワークスペースによってはその時間に予定を何も登録していないこともあるので、効率がいい方法とは言えません。

なんらかの方法で、対象となるデータを1カ所に集約したほうがよさそうです。

DynamoDBにデータを集約

私たちはDynamoDBにデータを集約することにしました。DynamoDBには予定日時、配信対象者、リマインダー送信方法（メール、チャット、Slack）など最小限の情報のみを保存しています。すべてのワークスペースのデータが集約されるため、以下の理由からDynamoDBを選択しています。

• ストレージの容量を意識する必要がない
• 負荷に応じてオートスケーリングする
• 読み込み書き込みが早い
• 正規化が必要となる複雑なデータの持ち方をする必要がない

DynamoDBではリレーショナル型のデータベースと違った視点での設計が必要です。DynamoDB を使用した設計とアーキテクチャの設計に関するベストプラクティスを参考にデータベースの設計を行っています。

運用中のサービスへの適用

サービスローンチ時であれば、リマインダーの送信件数は0からだんだんと増えていくので利用状況を見ながらパフォーマンスを改善できます。しかし2016年の時点ですでに5年ほどサービスを運用しており、新たな機能をすべてのユーザーに提供するとなると、どの程度のパフォーマンスが求められるのかわかりません。

推測するな、計測せよ

これについてはユーザーが登録している9:00の予定の数を計測することで予測を立てました。

2016年当時の数字ですが、9:00に2,800件ほどの予定が登録されていました。9:30や10:00の予定の数も調べましたが、9:00の予定が最も多いことがわかりました。

計測した結果から、

30秒以内に3,000件のリマインダーを配信できるようにする。

という目標を立てて開発を進めました。

配信フロー

リマインダー配信の全体的な流れはこのようになります。

それではそれぞれを詳しく見ていきましょう。

1. スケジュール登録

対象となるデータを1カ所に集約する必要があるため、予定登録時にRDSだけでなくDynamoDBにも予定を登録します。DynamoDBには予定日時、配信対象者、リマインダー送信方法（メール、チャット、Slack）など最小限の情報のみを入れるようにします。

2. Lambdaによる定期チェック

CloudWatch Eventsをトリガーにして定期的にLambdaからDynamoDBをチェックします。DynamoDBに登録されている、15分以内に開始する予定を抽出しSQSに渡します。

また、このタイミングでSimpleDBに保存されている顧客の契約情報をチェックしてリマインダー配信対象として正しいかどうかの判定をしています。

SQSには遅延キューというしくみがあり、DelaySeconds のパラメータ指定により最大15分間メッセージの配信を延期できます。

配信したいリマインダーの情報をあらかじめキューにためておいて、予定時間まで待たせておくようなイメージです。

リマインダー配信時の30秒に比べれば15分は非常に余裕のある時間のため、ここで最終的なデータの整合性チェックをしてリマインダー配信処理の負担を減らしています。

3. リマインダー配信

予定時刻になるとTomcat上で動いているDaemonがSQSを受け取り、リマインダーを配信します。この処理をいかに早く完了させられるかが重要になります。

SQSから送られてくる情報をもとに

• SimpleDBから顧客の契約情報を取得
• DynamoDBから配信対象者などの情報を改めて取得
• RDSから予定のタイトルや内容を取得

しています。これらのデータの整合性に問題がなければリマインダーを配信します。

Aipoでは複数のEC2インスタンスが稼働しており、それぞれのインスタンスでDaemonを1スレッドずつ実行することで並列処理をするようにしました。

パフォーマンス改善の道のり

最初のパフォーマンス

1Daemonあたりの30秒での処理件数は17件ほどでした。処理できる総数はインスタンスの台数に依存しますが、当時のインスタンス台数では30秒で100件までしか処理ができず、目標の 3% しかリマインダーの配信が間に合わないことがわかりました。

このままでは実用に耐えられません。予定が終わるころにリマインダーが届くことになってしまいます。

解決策

インスタンス1台あたりのDaemon実行数を増やす

さいわいにもインスタンスのスペックにはまだ余裕がありました。それぞれのインスタンスでDaemonの同時実行数を1スレッドから10スレッドまで増やしました。

Aipoではpropertiesに次のように書くことで同じクラスのDaemonを複数実行できます。

# デーモン名
daemon.entry=ReminderDaemon
# デーモンクラスへのパス
daemon.ReminderDaemon.classname=com.sample.ReminderDaemon
# 繰り返す間隔（秒）
daemon.ReminderDaemon.interval=1
# Servlet起動時に実行するか
daemon.ReminderDaemon.onstartup=true

daemon.entry=ReminderDaemon1
daemon.ReminderDaemon1.classname=com.sample.ReminderDaemon
daemon.ReminderDaemon1.interval=1
daemon.ReminderDaemon1.onstartup=true

複数のスレッドによる実行が手軽にできるのもJavaの強みです。

この対応により30秒で1,000件まで処理できるようになりました。目標が3,000件なのでさらに3倍のパフォーマンスにする必要があります。

1件にかかる処理を早くする

当初リマインダー配信のタイミングでRDS以外にDynamoDBとSimpleDBにアクセスして整合性をチェックした上でリマインダーの配信を行っていました。しかしWeb APIによるデータベースへのアクセスはどうしても時間がかかります。

SimpleDBとDynamoDBのチェックはSQSにキューを送信するタイミングでも行っているため、リマインダー配信時にはRDSのみにアクセスするよう変更しました。

• SimpleDBから顧客の契約情報を取得
• DynamoDBから配信対象者などの情報を改めて取得
• RDSから予定のタイトルや内容を取得

Lambdaの処理

SQSにメッセージを送る際にDynamoDBのデータをMessageAttributesに含めることで、Daemon側からDynamoDBへのアクセスが不要になります。

// item ： DynamoDB の item
// delay ： 遅延キューの時間（秒）
function reminderSQS(item, delay, callback) {
  var params = {
    MessageBody: item.id['S'],
    QueueUrl: 'https://sqs.ap-northeast-1.amazonaws.com/your/endpoint',
    DelaySeconds: delay,  // 遅延キューの時間
    MessageAttributes: {
      date: { // 予定日時
        DataType: 'String',
        StringValue: item.date['S']
      },
      type: { // リマインダー送信方法（メール、チャット、Slack）
        DataType: 'String',
        StringValue: item.type['SS'].join(':')
      }
    }
  };
  sqs.sendMessage(params, callback);
}

Daemonの処理

SQSのメッセージを受け取るときにはwithMessageAttributeNamesを指定することで、MessageAttributesを取得することが可能です。

public void run() {
    AmazonSQS sqs = SQS.getClient();

    ReceiveMessageResult receiveMessage =
        sqs
        .receiveMessage(
            new ReceiveMessageRequest("https://sqs.ap-northeast-1.amazonaws.com/your/endpoint")
            .withMaxNumberOfMessages(5) // 5件ずつメッセージを取得
            .withVisibilityTimeout(60 * 5)
            .withWaitTimeSeconds(20)
            .withMessageAttributeNames("All")); // MessageAttributesも取得

    List < Message > messages = receiveMessage.getMessages();
    for (Message message: messages) {
        String receiptHandle = message.getReceiptHandle();
        try {
            String key = message.getBody();
            Map < String, MessageAttributeValue > messageAttributes =
                message.getMessageAttributes();
            String date =
                messageAttributes.get("date").getStringValue();
            String type =
                messageAttributes.get("type").getStringValue();

            // TODO リマインダー送信処理

        } catch (Throwable t) {
            // TODO エラーハンドリング
        } finally {
            sqs
                .deleteMessage(
                    new DeleteMessageRequest("https://sqs.ap-northeast-1.amazonaws.com/your/endpoint", receiptHandle));
        }
    }
}

DynamoDBとSimpleDBにアクセスしないことで、1Daemonあたりの30秒に処理できる件数が17件から55件（3.2倍！）にまで改善されました。

まとめ

最終的に30秒で3,300件まで処理できるようになり、目標を達成することができました！

現在ではインスタンスの強化により、おおむね数秒以内に配信が完了しています。今後リマインダーの配信件数が増えた場合も、インスタンスの台数を増やすことで対応できるしくみを作ることができました。