Amazon SQS によるポイントツーポイントメッセージング

導入

ポイントツーポイントメッセージングパターンは、現代のWebおよびクラウドアーキテクチャで広く使用されている通信モデルです。これは、サーバーレス関数やマイクロサービスなどの異なるコンポーネント間の非同期的なやり取りを可能にするように設計されており、即時の応答を必要とせずにメッセージを交換できます。.

このパターンでは、メッセージを送信するコンポーネントはプロデューサー、メッセージを受信して処理するコンポーネントはコンシューマーと呼ばれます。プロデューサーとコンシューマーは、同じシステムに配置することも、異なるシステムに配置することも可能であるため、柔軟かつスケーラブルな通信アプローチとなります。.

電子メールが受信者に配信されるのと同様に、メッセージはプロデューサーから特定のコンシューマーに送信されます。これにより、複雑な分散システムにおいても効率的で信頼性の高い通信が可能になります。これは通常、プロデューサーがどのコンシューマーがメッセージを受信すべきかを正確に把握しているものの、プロデューサーが即時の応答を受け取る必要がないシナリオで使用されます。.

ポイントツーポイントメッセージングパターンは、コンポーネント間の通信と調整を効果的に促進し、最新の Web およびクラウドアーキテクチャの全体的なパフォーマンス、信頼性、およびスケーラビリティを向上させます。.

私たちが構築するもの

このステップバイステップのチュートリアルでは、2 つの AWS Lambda 関数と Amazon SQS キューを使用してこのパターンを実装します。.

このステップバイステップのチュートリアルでは、2 つの AWS Lambda 関数と Amazon SQS キューを使用して簡単な例を実装します。.

TypeScript と AWS クラウド開発キット (AWS CDK) を使用してサンプルを構築します。.

この例は次の 3 つのコンポーネントで構成されます。

消費者にメッセージを送信できるプロデューサー
プロデューサーからメッセージを受信できるコンシューマー。
プロデューサーとコンシューマー間の通信チャネルを作成するメッセージキュー。

このパターンの実装に加えて、インフラストラクチャ全体をコードとして定義できるAWSクラウド開発キット（CDK）の威力についても解説します。AWS CDKについて詳しく知りたい方は、AWS CDK開発者ガイドをご覧ください。.

このチュートリアルを完了すると、キューベースのポイントツーポイントメッセージングの個々のコンポーネントを完全に理解し、SQSを使用して2つのLambda関数間の非同期通信を実装し、インフラストラクチャ構築の実践的な経験を積むことができます。CDKでコードを作成する

しかし、コーディングを始める前に、非同期のポイントツーポイントメッセージングパターンの利点と欠点を簡単に見てみましょう。.

非同期ポイントツーポイントメッセージングパラダイムの利点と欠点

利点

疎結合：非同期のポイントツーポイントメッセージングパターンは、アプリケーション間の疎結合を促進し、緊密な統合を必要とせずにアプリケーションが独立して通信できるようにします。この柔軟性により、システム全体に影響を与えることなく、個々のコンポーネントの拡張や変更が容易になります。.
スケーラビリティ：このパターンでは、複数のコンシューマーアプリケーションを追加してワークロードを非同期的に処理できるため、効率的な水平スケーリングが可能になります。これにより、システムは大量の同時メッセージやリクエストをより効率的に処理できるようになります。.
信頼性: 非同期メッセージングでは、メッセージが配信または処理されない場合、再試行するか、後で処理するためにエラーキューに送信できるため、システムの信頼性が向上します。.
フォールトトレランス：非同期メッセージングは、メッセージのプロデューサーとコンシューマーを分離することでフォールトトレランスを実現します。アプリケーションまたはコンポーネントのいずれかがクラッシュした場合でも、メッセージは保存され、システムがオンラインに復帰した後に処理されるため、データの損失は発生しません。.

デメリット

複雑さ: 非同期のポイントツーポイントメッセージングパターンの実装は、他の統合パターンと比較して複雑になる可能性があり、追加のメッセージ処理ロジックが必要になります。.
メッセージの依存関係と重複の可能性：非同期メッセージングシステムでは、メッセージ間の依存関係を管理し、適切なメッセージ削除を確実に行うことが困難な場合があります。メッセージの順序付け、メッセージの重複、メッセージ処理の依存関係といった潜在的な問題に対処するには、慎重な設計と実装が必要です。.
レイテンシの増加：非同期メッセージングでは、メッセージの処理に時間がかかるため、メッセージの送信と応答の受信の間に遅延が発生します。このレイテンシはリアルタイムのインタラクションに影響を与える可能性があり、即時のフィードバックを必要とするアプリケーションには適さない場合があります。.

アーキテクチャ上の決定を行う際には、常にこれらのトレードオフを考慮し、特定の要件と制約に最適な通信パターンを選択することが重要です。多くの最新アプリケーションは、非同期のポイントツーポイントメッセージング、同期のリクエストレスポンス通信、イベントドリブン通信など、複数の統合パターンに依存しています。.

では、チュートリアルを始めて、AWS Lambda と Amazon SQS を使用してこのパターンを実装する方法を学びましょう。.

コーディング時のリソースコストに関する注意: このチュートリアルでは、最小限のリソースのみを使用します。これらのリソースはすべて、各アカウントの作成後の最初の 12 か月間は AWS によって提供される無料利用枠に含まれています。

数キロバイトのコードは、5 GB の無料ストレージを提供する Amazon S3 に保存されます。.
SQS を数回呼び出します。SQS では、1 か月あたり 100 万件の無料リクエストが提供されます。.
AWS Lambda で 2 つの関数を呼び出します。これも、1 か月あたり 100 万回の無料呼び出しを提供します。.

ステップバイステップガイドに従えば、無料利用枠を維持できます。また、このチュートリアルで作成したすべてのリソースを削除する方法については、最後にセクションを追加しました。.

前提条件

CLIとCDKがインストールされている場合は、インストールのバージョンが表示されます。インストールされていない場合は、コマンドが見つからないというエラーが表示されます。.

このチュートリアルを始める前に、次のものが必要です。

AWSアカウント
AWS クラウド開発キット (AWS CDK)

このチュートリアルでは、AWS CLI と AWS CDK のバージョン 2 以上が必要です。.

ターミナルで次のコマンドを実行すると、AWS CLI のバージョンを確認できます。

aws --version

次のコマンドを実行して AWS CDK のバージョンを確認します。

cdk --version

ステップ1 – CDKアプリケーションを作成する

1.1 – CDKプログラムを起動する

まず、初期設定を行い、プロジェクトをビルドするコードを CDK が自動的に生成するようにする必要があります。.

コマンドラインから、プロジェクト用のディレクトリを作成し、そのディレクトリに cd して、プロジェクトを初期化します。

mkdir point-to-point-example
cd point-to-point-example
cdk init app --language typescript

これにより、プロジェクトのメインファイル構造が作成され、必要なパッケージがインストールされます。.

ここで、お気に入りの IDE でプロジェクトを開き、生成されたプロジェクト、具体的には次の 2 つのファイルを確認するとよいでしょう。

point-to-point-example/
├── bin/
│ └── point-to-point-example.ts
├── lib/
│ └── point-to-point-example-stack.ts
└── ...

ファイル bin\ポイントツーポイント例.ts CDKプログラムのエントリポイント。このチュートリアルでは、このファイルを再度確認する必要はありません。.

ファイル lib\ポイントツーポイントの例スタック.ts プログラムのメインスタックが定義される場所です。このファイルで私たちはほとんどの時間を費やします。.

このチュートリアルを進めていくと、アプリケーションのインフラストラクチャをクラスに紹介します。 ポイントツーポイント例スタック このファイルに以下を追加します。エディターで開いて確認してください。

import * as cdk from 'aws-cdk-lib';
import { Construct } from 'constructs';
// import * as sqs from 'aws-cdk-lib/aws-sqs';
export class PointToPointExampleStack extends cdk.Stack {
constructor(scope: Construct, id: string, props?: cdk.StackProps) {
super(scope, id, props);
// The code that defines your stack goes here
// ...
}
}

1.2 – 環境のブートストラップ

AWS CDK アプリケーションを環境（特定の AWS アカウントとリージョン）に初めてデプロイする場合は、CDKToolkit スタックを使用して環境を起動する必要があります。このスタックは、実際のアプリケーションデプロイプロセスでテンプレートとアセットを保存するために使用する S3 バケットをデプロイします。.

環境をブートストラップするには、サンプルのポイントツーポイントディレクトリ内にいることを確認して、次のコマンドを実行します。

cdk bootstrap

これにはしばらく時間がかかりますが、コマンドラインに次の出力が表示されたら完了です。

✅ Environment aws://[account_id]/[region] bootstrapped.

1.3 – 初期スタックのデプロイ

スタックをデプロイするには、サンプルのポイントツーポイントディレクトリ内にいることを確認して、次のコマンドを実行します。

cdk deploy

このコマンドは、アプリケーションを AWS CloudFormation テンプレートにコンパイルし、AWS アカウントにデプロイします。.

スタックの詳細を開いて確認してみましょう。「リソース」タブを見ると、スタックの初期設定情報である「CDKMetadata」というリソース以外は何もデプロイされていないことがわかります。.

ステップ2 – SQSキューを作成する

すべての設定が完了したら、最初の実際のリソースである SQS キューを作成できます。.

2.1 – キューをスタックに追加する

まず、ファイル lib\ポイントツーポイントの例スタック.ts 開ける。.

CDKテンプレートのバージョンによっては、コメント内に aws-cdk-lib/aws-sqs を含むインポート文が既に追加されている場合があります。その場合は、import * as sqs from 'aws-cdk-lib/aws-sqs' の前のコメント記号 // を削除してください。そうでない場合は、ファイルの最初の3行が以下のようになるように、コメント記号を自分で追加してください。

import * as cdk from 'aws-cdk-lib';
import { Construct } from 'constructs';
import * as sqs from 'aws-cdk-lib/aws-sqs';

さて、授業でのコメントです。 ポイントツーポイント例スタック 次のコードを削除して置き換え、変更を保存します。

const queue = new sqs.Queue(this, 'Queue', {
queueName: 'MyQueue',
});

2.2 – キューをデプロイする

キューをデプロイするには、プロジェクトディレクトリで次のコマンドを実行します。

cdk deploy

これで、ターミナル出力でデプロイメントプロセスを監視できるようになりました。デプロイメントプロセスは 1 ～ 2 分後に完了するはずです。.

デプロイが完了したら、AWS マネジメントコンソールの SQS キューリストに移動し、SqsTutorialStack-Queue というキューを見つけます。

2.3 – デプロイ済みキューを確認する

AWS マネジメントコンソールの SQS ダッシュボードに移動し、キューのリストを確認してください。キューが見つからない場合は、正しいリージョンにいることを確認してください。.

ステップ3 – プロデューサー関数を作成し、キューに接続する

3.1 – ジェネレータ関数のソースコードを書く

Lambda関数にはJavaScriptを使用し、ソースコードをCDKアプリケーション内の専用フォルダに保存します。後ほどCDKスタックからこのフォルダを参照することで、コンテンツを自動的にパッケージ化し、アップロードしてAWSにデプロイできるようになります。.

プロジェクトディレクトリにいることを確認し、次のディレクトリとファイル自体を作成します。

mkdir lambda-src
mkdir lambda-src/producer
touch lambda-src/producer/send_message_to_sqs.js

プロジェクトのディレクトリ構造は次のようになります。

point-to-point-example/
├── bin/
│ └── point-to-point-example.ts
├── lambda-src/
│ └── producer/
│ └── send_message_to_sqs.js
├── lib/
│ └── point-to-point-example-stack.ts
└── ...

新しいファイルが作成されました。 lambda-src/producer/send_message_to_sqs.js エディターで開き、次のコードを追加します。

const sqs = require("@aws-sdk/client-sqs");
// Create an Amazon SQS service client
const client = new sqs.SQSClient();
exports.handler = async (event, context) => {
// Create a message, including the function's name from
// its execution context
const myMessage = 'Hello World from ' + context.functionName
// Get the queue URL from the execution environment
const queueUrl = process.env.SQSQueueUrl
// Create a SendMessageCommand containing the message
// and the queue URL
const command = new sqs.SendMessageCommand({
MessageBody: JSON.stringify({ message: myMessage }),
QueueUrl: queueUrl
});
// Send the message and return the result 
const result = await client.send(command);
return result;
}

これは、SQS キューにメッセージを送信し、結果を返す単純な Lambda 関数ハンドラーです。.

この関数をビルドする際にどのキューを使用するか必ずしも分かっているとは限らないため、あるいは関数コードを変更せずに後でキューを変更したい場合もあるため、ランタイムからキューのURLを取得します。これにより、CDKスタックにLambda関数をデプロイする際に動的にURLを定義できます。.

3.2 – プロデューサー関数をスタックに追加する

AWSで実際のLambda関数を作成するには、スタックに追加する必要があります。そのためには、lib/point-to-point-example-stack.tsを開いてください。.

まず、ファイルの先頭に別のインポートステートメントを追加します。

import * as lambda from 'aws-cdk-lib/aws-lambda';

次に、コンストラクター内のキューの下に次のスニペットを追加します。

const producerFunction = new lambda.Function(this, 'Producer', {
functionName: 'Producer',
runtime: lambda.Runtime.NODEJS_18_X,
code: lambda.Code.fromAsset('lambda-src/producer'),
handler: 'send_message_to_sqs.handler',
environment: {
SQSQueueUrl: queue.queueUrl,
}
});

これにより、関数、パッケージ、および lambda-src/producer ディレクトリの内容が作成され、呼び出されたときに send_message_to_sqs.js で定義されたハンドラーを使用するように関数が構成されます。.

また、SQSQueueUrl 環境変数を SQS キューの実際の URL に動的に設定し、process.env.SQSQueueUrl を使用して関数内からアクセスできるようにします。.

3.3 – プロデューサーのパフォーマンスのライセンス

各 Lambda 関数にはいくつかの基本的な IAM 権限が必要ですが、CDK はデフォルトで基本実行ロールを使用してこれを提供します。.

ただし、プロデューサー関数はキューにメッセージを送信するために追加の権限が必要です。幸いなことに、CDKではIAMポリシーを手動で管理する必要はありません。Lambda関数でSQSキューの grantSendMessages メソッドを使用できます。.

これを行うには、producerFunction の下に次の行を追加します。

queue.grantSendMessages(producerFunction);

3.4 – プロデューサー関数のデプロイ

関数をそのソースコードとともにデプロイするには、プロジェクトディレクトリで次のコマンドを実行します。

cdk deploy

これにより、新しいリソースまたは変更されたリソースの展開が準備されます。.

スタックへの変更で IAM 権限の作成または変更が必要な場合は、セキュリティ上の理由から、CDK によってこれらの変更を再確認するように求められます。.

下の画像では、CDK が IAM 権限に影響する 3 つの変更を加えようとしていることがわかります。

Lambda サービス (lambda.amazon.aws) が Producer 関数にアタッチされた実行ロールを引き受けることを許可します。.
必要な権限を直接追加して、エグゼクティブロールが SQS キューにアクセスしてメッセージを送信できるようにします。.
実行ロールに初期権限を持つ管理ポリシーを追加します。.

確認のために y と入力すると、CDK は新しい Lambda 関数とその IAM ロールをデプロイします。.

3.5 – プロデューサーのパフォーマンスをテストする

関数を実行した後、AWS Lambda ダッシュボードの関数のリストに移動します。.

関数名はスタックに定義したものと同じです: Producer。関数が見つからない場合は、正しいリージョンにいることを確認してください。.

関数名をクリックすると詳細が開き、ソースコードセクションに移動します。.

左側のファイルエクスプローラーフォルダーには、send_message_to_sqs.js というファイルが含まれています。.

ファイル名をダブルクリックしてソースを開くと、CDK プロジェクトの lambda-src/producer フォルダーに作成したファイルの正確なコピーであることがわかります。.

それでは、テストイベントを設定しましょう。.

これを行うには、「テスト」ボタンをクリックします。まだテストイベントを設定していない場合は、「テスト設定」ページが開きます。イベントの名前（例：「test-event」）を入力し、「保存」をクリックします。.

テストイベントが設定されたら、もう一度 [テスト] をクリックすると、Lambda 関数が直接呼び出されます。.

これにより、応答を含む実行結果タブが作成されます。

{
"$metadata": {
"httpStatusCode": 200,
"requestId": "cb033b2a-1234-5678-9012-66188359d280",
"attempts": 1,
"totalRetryDelay": 0
},
"MD5OfMessageBody": "b5357af0c1c816b2d41275537cc0d1af",
"MessageId": "4a76e538-1234-5678-9012-749f0f4b9294"
}

Lambda 関数のコードスニペットを覚えていますか?

const result = await client.send(command);
return result;

上記の実行結果の JSON オブジェクトは、SQS が client.send(command) への応答として返したものとまったく同じで、キューがメッセージを正常に受信したことを意味する「httpStatusCode」: 200 などの興味深い情報が含まれています。.

3.6 – SQSからメッセージを手動で取得する

現在、キューはコンシューマーに接続されていないため、メッセージは受信されません。ただし、手動でメッセージを取得することは可能です。.

SQS ダッシュボードに移動し、キューをクリックして詳細ページを開き、「メッセージの送受信」ボタンをクリックします。.

次のページで、[メッセージの受信] セクションに移動し、[メッセージのポーリング] ボタンをクリックします。これにより、テスト時にビルダー関数が送信したメッセージを含む、現在送信中のすべてのメッセージが表示されます。

メッセージIDをクリックして開きます。本文には以下の内容が含まれています。

{"message":"Hello World from Producer"}

おめでとうございます！SQS キューにメッセージを送信する Lambda 関数のデプロイとテストに成功しました。.

次に、コンシューマー関数を作成し、キューに新しいメッセージが到着するたびに自動的に呼び出されるように設定します。.

ステップ4 – Consumer関数を作成し、キューに接続する

4.1 – コンシューマー関数のソースコードを書く

コンシューマー関数コードは lambda-src 内の独自のフォルダーに保存されるため、CDK はそれを自動的にパッケージ化し、新しい Lambda 関数にアップロードできます。.

プロジェクトディレクトリ (bin/ または lib/ ではない) にいることを確認し、ソースファイルを含む次のディレクトリを作成します。

mkdir lambda-src/consumer
touch lambda-src/consumer/receive_message_from_sqs.js

エディターで lambda-src/consumer/receive_message_from_sqs.js を開き、次のコードを入力します。

exports.handler = async (event) => {
// Retrieve the message from the event and log it to the console
const message = event.Records[0].body;
console.log(message);
}

これは、SQS メッセージを開いてレポートに出力する非常にシンプルな Lambda 関数ハンドラーです。.

4.2 – スタックにConsumer関数を追加する

AWSで実際のLambda関数を作成するには、スタックに追加する必要があります。そのためには、lib/point-to-point-example-stack.tsを開いてください。.

プロデューサー内の queue.grantSendMessages(producerFunction) の下に次のスニペットを追加します。

 const consumerFunction = new lambda.Function(this, 'Consumer', {
functionName: 'Consumer',
runtime: lambda.Runtime.NODEJS_18_X,
code: lambda.Code.fromAsset('lambda-src/consumer'),
handler: 'receive_message_from_sqs.handler'
});

このコンシューマー関数は、lambda-src/consumer ディレクトリのパッケージとコンテンツを作成し、呼び出されたときに receive_message_from_sqs.js で定義されたハンドラーを使用するように関数を構成します。.

4.3 – SQSキューをイベントソースとしてコンシューマーに追加する

キューに新しいメッセージがあるたびにこの関数が実行されるようにします。そのためには、キューを関数のイベントソースとして追加し、メッセージを処理するために必要なすべての権限を関数に付与する必要があります。.

まず、ファイルの先頭に別のインポートステートメントを追加します。

import { SqsEventSource } from 'aws-cdk-lib/aws-lambda-event-sources';

次に、customersFunction 定義の下に次の 2 行を追加します。

consumerFunction.addEventSource(new SqsEventSource(queue));
queue.grantConsumeMessages(consumerFunction);

4.4 – コンシューマー関数のデプロイ

関数をそのソースコードとともにデプロイするには、プロジェクトディレクトリで次のコマンドを実行します。

cdk deploy

これにより、新しいリソースまたは変更されたリソースのデプロイメントが準備され、再度、新しい IAM 権限の再検証が求められます。.

確認のために y と入力すると、CDK は新しい Lambda 関数とその実行ロールを実行し、SQS キューにアタッチします。.

デプロイが完了すると、すべての準備が完了し、接続されます。動作するか確認してみましょう。

ステップ5 – アプリケーションのエンドツーエンドテスト

アプリケーションをテストするために、プロデューサー関数を手動で呼び出してキューに別のメッセージを送信します。次に、コンシューマー関数の実行ログをチェックして、メッセージが自動的に起動され処理されたかどうかを確認します。.

5.1 – プロデューサー関数を呼び出す

3.5 – プロデューサー関数のテストに記載されている手順に従って、プロデューサー関数を再度呼び出します。以下に概要を示します。

AWS Lambda ダッシュボードの関数リストに移動します。.
名前をクリックして、プロデューサー機能を開きます。.
ソースコードセクションへ移動します。.
[テスト]ボタンをクリックします。.
関数が正常に実行されることを確認します。.

5.2 – 消費者パフォーマンスレポートを確認する

AWS Lambda ダッシュボードの関数リストに戻り、今度は Consumer 関数を開きます。.

下にスクロールして、「モニター」というタブを開きます。このタブ内で「ログ」タブを開き、「最近の呼び出し」を確認します。ここで、関数の最後の LogStream へのリンクが見つかるはずです。.

その後、「ログフロー」リンクをクリックします。ログエントリ自体が表示されますが、形式に慣れていないと少し分かりにくいかもしれません。.

これら 2 つの中間の、START RequestId: … と END Request ID: … で始まるエントリを探すだけです。.

矢印をクリックするとエントリが展開され、元のメッセージが表示されます。

結果

おめでとうございます！SQS を使用して 2 つの Lambda 関数間で非同期のポイントツーポイントメッセージングを設定するために必要なものがすべて作成されました。.

SQS では、メッセージの分類、メッセージの順序の保証、無効なメッセージのデッドラインキュー (DLQ) への自動送信など、他にも多くの機能があります。詳しくは、Amazon SQS 開発者ガイドをご覧ください。