AWS SDK for Java 2.x で Amazon SQS の自動リクエストバッチ処理を使用する - AWS SDK for Java 2.x
前提条件をチェックするバッチマネージャーを作成するメッセージを送信するメッセージの可視性タイムアウトを変更するメッセージの削除メッセージを受信する構成設定

AWS SDK for Java 2.x で Amazon SQS の自動リクエストバッチ処理を使用する

Amazon SQS の自動リクエストバッチ処理 API は、SQS オペレーションのリクエストをバッチ処理およびバッファリングする効率的な方法を提供する高レベルライブラリです。バッチ処理 API を使用すると、SQS へのリクエストの数を減らすことができるため、スループットが向上し、コストが最小限に抑えられます。

バッチ処理 API メソッドは、SqsAsyncClient メソッド (sendMessage、changeMessageVisibility、deleteMessage、receiveMessage) と一致するため、最小限の変更でドロップイン置換としてバッチ処理 API を使用できます。

このトピックでは、Amazon SQS の自動リクエストバッチ処理 API を設定して使用する方法の概要を説明します。

前提条件をチェックする

バッチ処理 API にアクセスするには、SDK for Java 2.x のバージョン 2.28.0 以降を使用する必要があります。Maven pom.xml には、少なくとも次の要素が含まれている必要があります。

<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>software.amazon.awssdk</groupId>
            <artifactId>bom</artifactId>
            <version>2.28.231</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>
<dependencies>
    <dependency>
        <groupId>software.amazon.awssdk</groupId>
        <artifactId>sqs</artifactId>
    </dependency>
</dependencies>

1 最新バージョン

バッチマネージャーを作成する

自動リクエストバッチ処理 API は、SqsAsyncBatchManager インターフェイスによって実装されます。マネージャーのインスタンスは、いくつかの方法で作成できます。

SqsAsyncClient を使用したデフォルト設定

バッチマネージャーを作成する最も簡単な方法は、既存の SqsAsyncClient インスタンスで batchManager ファクトリメソッドを呼び出すことです。次のスニペットは、シンプルなアプローチを示しています。

SqsAsyncClient asyncClient = SqsAsyncClient.create();
SqsAsyncBatchManager sqsAsyncBatchManager = asyncClient.batchManager();

この方法を使用する場合、SqsAsyncBatchManager インスタンスは SqsAsyncBatchManager の設定を上書きする セクションの表に示されているデフォルト値を使用します。さらに、SqsAsyncBatchManager インスタンスは、作成された SqsAsyncClient インスタンスの ExecutorService を使用します。

SqsAsyncBatchManager.Builder を使用したカスタム設定

より高度なユースケースでは、SqsAsyncBatchManager.Builder を使用してバッチマネージャーをカスタマイズできます。このアプローチを使用して SqsAsyncBatchManager インスタンスを作成すると、バッチ処理の動作をファインチューニングできます。次のスニペットは、ビルダーを使用してバッチ処理の動作をカスタマイズする方法の例を示しています。

SqsAsyncBatchManager batchManager = SqsAsyncBatchManager.builder()
    .client(SqsAsyncClient.create())
    .scheduledExecutor(Executors.newScheduledThreadPool(5))
    .overrideConfiguration(b -> b
        .receiveMessageMinWaitDuration(Duration.ofSeconds(10))
        .receiveMessageVisibilityTimeout(Duration.ofSeconds(1))
        .receiveMessageAttributeNames(Collections.singletonList("*"))
        .receiveMessageSystemAttributeNames(Collections.singletonList(MessageSystemAttributeName.ALL)))
    .build();

このアプローチを使用すると、 SqsAsyncBatchManager の設定を上書きする セクションの表に表示される BatchOverrideConfiguration オブジェクトの設定を調整できます。このアプローチを使用して、バッチマネージャーにカスタム ScheduledExecutorService を指定することもできます。

メッセージを送信する

バッチマネージャーでメッセージを送信するには、 SqsAsyncBatchManager#sendMessage メソッドを使用します。SDK はリクエストをバッファリングし、 maxBatchSize または sendRequestFrequency の値に達するとバッチとして送信します。

次の例では、sendMessage リクエストの直後に別のリクエストが発生しています。この場合、SDK は両方のメッセージを 1 つのバッチで送信します。

// Sending the first message
CompletableFuture<SendMessageResponse> futureOne = 
    sqsAsyncBatchManager.sendMessage(r -> r.messageBody("One").queueUrl("queue"));

// Sending the second message
CompletableFuture<SendMessageResponse> futureTwo = 
    sqsAsyncBatchManager.sendMessage(r -> r.messageBody("Two").queueUrl("queue"));

// Waiting for both futures to complete and retrieving the responses
SendMessageResponse messageOne = futureOne.join();
SendMessageResponse messageTwo = futureTwo.join();

メッセージの可視性タイムアウトを変更する

SqsAsyncBatchManager#changeMessageVisibility メソッドを使用して、メッセージの可視性タイムアウトをバッチで変更できます。SDK はリクエストをバッファリングし、 maxBatchSize または sendRequestFrequency の値に達するとバッチとして送信します。

次の例は、changeMessageVisibility メソッドを呼び出す方法を示しています。

CompletableFuture<ChangeMessageVisibilityResponse> futureOne =
    sqsAsyncBatchManager.changeMessageVisibility(r -> 
        r.receiptHandle("receiptHandle")
         .queueUrl("queue"));
ChangeMessageVisibilityResponse response = futureOne.join();

メッセージの削除

SqsAsyncBatchManager#deleteMessage メソッドを使用して、メッセージをバッチで削除できます。SDK はリクエストをバッファリングし、 maxBatchSize または sendRequestFrequency の値に達するとバッチとして送信します。

次の例は、 deleteMessage メソッドを呼び出す方法を示しています。

CompletableFuture<DeleteMessageResponse> futureOne = 
    sqsAsyncBatchManager.deleteMessage(r -> 
        r.receiptHandle("receiptHandle")
         .queueUrl("queue"));
DeleteMessageResponse response = futureOne.join();

メッセージを受信する

デフォルト設定を使用する

アプリケーションで SqsAsyncBatchManager#receiveMessage メソッドをポーリングすると、バッチマネージャーは内部バッファからメッセージを取得し、SDK は自動的にバックグラウンドで更新します。

次の例は、receiveMessage メソッドを呼び出す方法を示しています。

CompletableFuture<ReceiveMessageResponse> responseFuture = 
    sqsAsyncBatchManager.receiveMessage(r -> r.queueUrl("queueUrl"));

カスタム設定を使用する

例えばカスタム待機時間を設定し、取得するメッセージの数を指定してリクエストをさらにカスタマイズする場合は、次の例に示すようにリクエストをカスタマイズできます。

CompletableFuture<ReceiveMessageResponse> response = 
    sqsAsyncBatchManager.receiveMessage(r -> 
        r.queueUrl("queueUrl")
         .waitTimeSeconds(5)
         .visibilityTimeout(20));
注記

次のいずれかのパラメータを含む ReceiveMessageRequestで receiveMessage を呼び出すと、SDK はバッチマネージャーをバイパスし、通常の非同期 receiveMessage リクエストを送信します。

  • messageAttributeNames

  • messageSystemAttributeNames

  • messageSystemAttributeNamesWithStrings

  • overrideConfiguration

SqsAsyncBatchManager の設定を上書きする

SqsAsyncBatchManager インスタンスを作成するときに、次の設定を調整できます。BatchOverrideConfiguration.Builder では、以下の設定が可能です。

設定 説明 デフォルト値
maxBatchSize SendMessageBatchRequest、ChangeMessageVisibilityBatchRequest、または DeleteMessageBatchRequest ごとのバッチあたりのリクエストの最大数。最大値は 10 です。 10
sendRequestFrequency

すでに maxBatchSize に達していない場合の、バッチを送信するまでの時間。値を大きくするとリクエストは減少しますが、レイテンシーは増加する可能性があります。

 200 ms
receiveMessageVisibilityTimeout メッセージの可視性タイムアウト。設定しない場合、キューのデフォルトが使用されます。 キューのデフォルト
receiveMessageMinWaitDuration receiveMessage リクエストの最小待機時間。CPU の無駄を防ぐため、0 には設定しないでください。 50 ms
receiveMessageSystemAttributeNames receiveMessage 呼び出しをリクエストするシステム属性名のリスト。 なし
receiveMessageAttributeNames receiveMessage 呼び出しをリクエストする属性名のリスト。 なし