AWS SDK for Java 2.x - メッセージング (SQS & SNS)

概要

AWS SDK for Java 2.x を使用した SQS キューと SNS トピックのパターンを提供します。クライアントセットアップ、キュー管理、メッセージ操作、サブスクリプション、Spring Boot 統合に対応しています。

使用する場面

• メッセージバッファリング用の SQS キュー (標準または FIFO) のセットアップ
• SNS トピックとサブスクリプションを使用した pub/sub の実装
• ロングポーリングでの SQS キューからのメッセージ処理
• エラーハンドリング用のデッドレターキュー (DLQ) の構成
• AWS メッセージングと Spring Boot アプリケーションの統合
• SQS/SNS を使用したイベント駆動アーキテクチャの構築

例

クイックセットアップ

依存関係:

<dependency>
    <groupId>software.amazon.awssdk</groupId>
    <artifactId>sqs</artifactId>
</dependency>
<dependency>
    <groupId>software.amazon.awssdk</groupId>
    <artifactId>sns</artifactId>
</dependency>

クライアント設定:

SqsClient sqsClient = SqsClient.builder()
    .region(Region.US_EAST_1)
    .credentialsProvider(DefaultCredentialsProvider.create())
    .build();

SnsClient snsClient = SnsClient.builder()
    .region(Region.US_EAST_1)
    .build();

SQS 操作

キューの作成とメッセージの送信:

String queueUrl = sqsClient.createQueue(CreateQueueRequest.builder()
    .queueName("my-queue")
    .build()).queueUrl();

String messageId = sqsClient.sendMessage(SendMessageRequest.builder()
    .queueUrl(queueUrl)
    .messageBody("Hello, SQS!")
    .build()).messageId();

メッセージの受信と削除:

ReceiveMessageResponse response = sqsClient.receiveMessage(ReceiveMessageRequest.builder()
    .queueUrl(queueUrl)
    .maxNumberOfMessages(10)
    .waitTimeSeconds(20)
    .build());

response.messages().forEach(message -> {
    processMessage(message.body());
    sqsClient.deleteMessage(DeleteMessageRequest.builder()
        .queueUrl(queueUrl)
        .receiptHandle(message.receiptHandle())
        .build());
});

FIFO キュー:

Map<QueueAttributeName, String> attributes = Map.of(
    QueueAttributeName.FIFO_QUEUE, "true",
    QueueAttributeName.CONTENT_BASED_DEDUPLICATION, "true"
);

String fifoQueueUrl = sqsClient.createQueue(CreateQueueRequest.builder()
    .queueName("my-queue.fifo")
    .attributes(attributes)
    .build()).queueUrl();

sqsClient.sendMessage(SendMessageRequest.builder()
    .queueUrl(fifoQueueUrl)
    .messageBody("Order #12345")
    .messageGroupId("orders")
    .messageDeduplicationId(UUID.randomUUID().toString())
    .build());

SNS 操作

トピックの作成とパブリッシュ:

String topicArn = snsClient.createTopic(CreateTopicRequest.builder()
    .name("my-topic")
    .build()).topicArn();

snsClient.publish(PublishRequest.builder()
    .topicArn(topicArn)
    .subject("Test Notification")
    .message("Hello, SNS!")
    .build());

SNS から SQS へのサブスクリプション:

String queueArn = sqsClient.getQueueAttributes(GetQueueAttributesRequest.builder()
    .queueUrl(queueUrl)
    .attributeNames(QueueAttributeName.QUEUE_ARN)
    .build()).attributes().get(QueueAttributeName.QUEUE_ARN);

snsClient.subscribe(SubscribeRequest.builder()
    .protocol("sqs")
    .endpoint(queueArn)
    .topicArn(topicArn)
    .build());

Spring Boot 統合

@Service
@RequiredArgsConstructor
public class OrderNotificationService {
    private final SnsClient snsClient;
    private final ObjectMapper objectMapper;

    @Value("${aws.sns.order-topic-arn}")
    private String orderTopicArn;

    public void sendOrderNotification(Order order) throws JsonProcessingException {
        snsClient.publish(PublishRequest.builder()
            .topicArn(orderTopicArn)
            .subject("New Order Received")
            .message(objectMapper.writeValueAsString(order))
            .messageAttributes(Map.of(
                "orderType", MessageAttributeValue.builder()
                    .dataType("String")
                    .stringValue(order.getType())
                    .build()))
            .build());
    }
}

手順

メッセージ処理の実装 (検証付き)

1. キュー/トピックの作成 - 適切な設定で実施
2. メッセージの送信 - messageId が返されることを検証
3. メッセージの受信 - ロングポーリング (waitTimeSeconds: 20) で実施
4. メッセージの処理 - 処理前にペイロードを検証
5. メッセージの削除 - 処理成功後のみ削除、削除レスポンスを確認
6. DLQ の確認 - redrivePolicy を使用して定期的に失敗メッセージを確認
7. 配信の検証 - CloudWatch NumberOfMessagesSent メトリクスを監視

検証チェックリスト:

// 送信後
if (messageId == null || messageId.isEmpty()) {
    throw new MessagingException("Message send failed - no messageId returned");
}

// 受信後
if (response.messages().isEmpty()) {
    log.debug("No messages available - normal with long polling");
}

// 削除後
if (!deleteResponse.sdkHttpResponse().isSuccessful()) {
    throw new MessagingException("Message deletion failed");
}

認証情報のセットアップ

export AWS_ACCESS_KEY_ID=your-access-key
export AWS_SECRET_ACCESS_KEY=your-secret-key
export AWS_REGION=us-east-1

監視とデバッグ

• CloudWatch メトリクス: ApproximateNumberOfMessages、NumberOfMessagesSent、NumberOfMessagesReceived
• SDK ロギング有効化: software.amazon.awssdk を DEBUG レベルで設定
• X-Ray による分散トレーシング利用

ベストプラクティス

SQS:

• ロングポーリング (20～40 秒) を使用して空のレスポンスとコストを削減
• 処理成功後は必ずメッセージを削除
• 重複処理対応のためべき等処理を実装
• 失敗メッセージ用に DLQ (redrivePolicy) を構成
• 順序が重要な場合は FIFO キューを使用 (300 メッセージ/秒の制限あり)

SNS:

• フィルターポリシーで不要な配信を削減
• メッセージを 256KB 以下に保つ
• 指数バックオフによる再試行を実装
• NumberOfNotificationFailed メトリクスを監視

全般:

• 静的認証情報ではなく IAM ロールを使用
• クライアントを再利用 (スレッドセーフ)
• LocalStack または Testcontainers でテスト

詳細なリファレンス

• references/detailed-sqs-operations.md
• references/detailed-sns-operations.md
• references/spring-boot-integration.md
• references/aws-official-documentation.md

制約と警告

• メッセージサイズ: SQS と SNS の最大値は 256KB
• 可視性タイムアウト: 削除されないメッセージはタイムアウト後に再び表示されます。処理後は必ず削除してください
• 入力検証: メッセージボディを処理前にサニタイズしてください。メッセージに信頼できないペイロードが含まれる可能性があります
• FIFO ネーミング: 名前の末尾は .fifo である必要があります
• FIFO スループット: キュー単位で 300 メッセージ/秒 (高スループット用にはパーティショニングを使用)
• メッセージ保持期間: SQS はメッセージを最大 14 日間保持
• DLQ 必須: メッセージ損失を防ぐためにデッドレターキューを構成してください
• リージョン固有: SQS キューはリージョン固有です。クロスリージョンには SNS が必要です