AWS IoT Core のデータを SQS キューに転送して EC2 から AWS CLI で読み出す手順
目次
はじめに
AWS IoT Core でセンサーデータを受信できるようになった後、次に課題になるのが「受け取ったデータをどのサービスに渡して処理するか」です。IoT Core にはルール機能があり、受信したデータを各種 AWS サービスへ自動転送できます。
本記事では、IoT Core で受信したセンサーデータを Amazon SQS（Simple Queue Service） のキューに積み、EC2 から AWS CLI でそのデータを読み出す手順を説明します。SQS を中継することで、データの送受信のタイミングをずらしたり、複数のコンシューマーが同じデータを処理したりといった構成が取れるようになります。AWS サービス間の疎結合な連携を学びたい方の参考になれば幸いです。
なお、AWS IoT Core の初期設定（証明書・ポリシーの作成・Raspberry Pi からの MQTT 接続）については以下の記事を参照してください。
設定の全体像と作業フロー
今回の構成は IoT Core → SQS → EC2 というデータの流れです。設定の手順は以下のとおりです。
- SQS でキューを作成する
- IoT Core のルールを作成して SQS にデータを転送する
動作確認では以下を実施します。
- センサーデータを送信して SQS キューにメッセージが積まれていることを確認する
- EC2 から AWS CLI でキューのメッセージを読み出す
AWS IoT Core と SQS を連携させる2ステップの設定
1. SQS 標準キューを作成する
課題：IoT Core が転送先として使えるキューを事前に用意する必要がある。
解決策：SQS の標準キューを作成し、メッセージ保持期間だけ調整する。
AWS コンソールの SQS サービスページに移動し、「キューの作成」をクリックします。
キューの種類は標準キューを選択します。設定はほぼデフォルトのまま進めますが、メッセージの保持期間だけ用途に合わせて変更しておくことをおすすめします。
今回は my-iot-queue という名前でキューを作成しました。これで SQS 側の準備は完了です。
2. IoT Core ルールで SQS へデータを転送する
課題：IoT Core が受信した MQTT メッセージを SQS に自動転送する仕組みが必要。
解決策：IoT Core の「ルール」機能で転送先とフィルター条件（MQTT トピック）を設定する。
AWS IoT Core の「ルール」タブから「作成」をクリックします。
まずルールの名前と説明を入力します。
続いてルールクエリステートメントを設定します。ここでは SQL 形式でフィルター条件を記述し、どの MQTT トピックのデータを転送するかを指定します。MQTT トピックを対象にする場合は FROM の後にトピック名を指定してください。以下は sensor/001 トピックを対象にした例です。
次に「アクションの追加」をクリックして転送先を設定します。
転送先の選択肢として各種 AWS サービスが表示されます。この一覧を見るだけで、IoT Core がいかに多くのサービスと簡単に連携できるかがわかります。今回は「SQS キューにメッセージを送信する」を選択し、「アクションの設定」をクリックします。
アクションの設定画面では、先ほど作成した SQS キューを指定し、IAM ロールを選択します。適切なロールがない場合は「ロールの作成」をクリックすると、IoT Core が SQS にアクセスするためのロールを名前を付けるだけで自動生成してくれます。設定が完了したら「アクションの追加」をクリックします。
アクションが追加されたことを確認したら、画面下部の「ルールの作成」をクリックして完了です。
動作確認1：センサーデータが SQS キューに積まれることを確認する
センサーデータを AWS IoT Core に向けて送信した後、SQS のコンソールで対象キューの「利用可能なメッセージ数」が 0 より大きくなっていることを確認します。以下の例では MQTT で 1 件送信した状態で、メッセージが 1 件表示されています。
これで IoT Core → SQS 間のデータ転送が正常に動作していることを確認できました。
動作確認2：EC2 の AWS CLI で SQS キューからメッセージを読み出す
課題：EC2 から SQS にアクセスするには適切な権限が必要。
解決策：EC2 に SQS アクセス権限を持つ IAM ロールをアタッチする。
IAM ロールの作成・アタッチ方法は以下の記事を参照してください。
EC2（Amazon Linux）に SSH 接続し、AWS CLI でキューの一覧を確認します。エラーになる場合は IAM ロールのアタッチを確認してください。
aws sqs list-queues
作成したキューの URL が確認できたら、次のコマンドでメッセージを受信します。{QueueUrls} には上記コマンドで取得した URL を指定してください。
aws sqs receive-message --queue-url {QueueUrls}
レスポンスの Body フィールドにセンサーデータが含まれていることが確認できます。
メッセージを受信した直後は「処理中（In Flight）」状態となり、他のコンシューマーからは読み出せません。この不可視状態が続く時間は「デフォルトの可視性タイムアウト」（デフォルト 30 秒）で制御されます。タイムアウト後はメッセージが再び「利用可能」状態に戻るため、明示的に削除するまではキューにメッセージが残り続けます。
処理が完了したメッセージは以下のコマンドで削除します。{QueueUrls} はキューの URL、{ReceiptHandle} は receive-message のレスポンスに含まれる値を指定してください。受信者本人のみが持つ ReceiptHandle を使って削除する設計になっており、他のコンシューマーが誤って削除する事故を防ぐ仕組みです。
aws sqs delete-message --queue-url {QueueUrls} --receipt-handle {ReceiptHandle}
まとめ：AWS IoT Core と SQS の連携で実現できること
今回の記事で紹介したポイントをまとめます。
- AWS IoT Core のルール機能を使えば、MQTT トピック単位で転送先サービスを柔軟に設定できる
- SQS を中継することで、センサーデータの送信と処理のタイミングを非同期・疎結合にできる
- EC2 に適切な IAM ロールを付与すれば、AWS CLI だけでキューの閲覧・受信・削除が完結する
- メッセージの可視性タイムアウトを活用することで、処理失敗時の再配信制御も実現できる