aws-sdk-java-v2-messaging

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. redrivePolicy を使用して、失敗したメッセージがないか DLQ を定期的に確認します
7. 配信を検証します。CloudWatch の NumberOfMessagesSent メトリクスを監視してください

検証チェックリスト：

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

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

// After delete
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 msg/sec）

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 msg/sec です（スループットを上げるにはパーティショニングを使用します）
• メッセージの保持期間：SQS がメッセージを保持する期間は最長 14 日です
• DLQ が必須：メッセージの損失を防ぐため、デッドレターキューを設定してください
• リージョン固有：SQS キューはリージョン固有です。リージョン間で利用するには SNS が必要です