翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

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

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

バッチAPIメソッドは、sendMessage、、 changeMessageVisibilitySqsAsyncClientメソッドと一致するためdeleteMessage、最小限の変更でドロップイン代替APIとしてバッチreceiveMessageを使用できます。

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

前提条件をチェックする

for SDK Java 2.x のバージョン 2.28.0 以降を使用して、バッチ処理 にアクセスする必要がありますAPI。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>

¹ 最新バージョン

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

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

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

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

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

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

1. を使用したカスタム設定 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リクエストの直後のリクエストを示しています。この場合、 は両方のメッセージを 1 つのバッチでSDK送信します。

// 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));

の設定を上書きする SqsAsyncBatchManager

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