Mengaktifkan buffering sisi klien dan batching permintaan dengan Amazon SQS

AWS SDK untuk JavaTermasuk AmazonSQSBufferedAsyncClient yang mengakses Amazon SQS. Klien ini memungkinkan batching permintaan sederhana menggunakan buffering sisi klien. Panggilan yang dilakukan dari klien pertama kali di-buffer dan kemudian dikirim sebagai permintaan batch ke Amazon SQS.

Buffering sisi klien memungkinkan hingga 10 permintaan untuk di-buffer dan dikirim sebagai permintaan batch, mengurangi biaya penggunaan Amazon SQS dan mengurangi jumlah permintaan yang dikirim. AmazonSQSBufferedAsyncClientbuffer panggilan sinkron dan asinkron. Permintaan batch dan dukungan untuk polling panjang juga dapat membantu meningkatkan throughput. Untuk informasi selengkapnya, lihat Meningkatkan throughput menggunakan penskalaan horizontal dan batching aksi dengan Amazon SQS.

Karena AmazonSQSBufferedAsyncClient mengimplementasikan antarmuka yang sama sepertiAmazonSQSAsyncClient, migrasi dari AmazonSQSAsyncClient ke AmazonSQSBufferedAsyncClient biasanya hanya memerlukan sedikit perubahan pada kode Anda yang ada.

catatan

Klien Asinkron Buffered Amazon SQS saat ini tidak mendukung antrian FIFO.

Menggunakan Amazon SQSBuffered AsyncClient

Sebelum memulai, selesaikan langkah-langkah di Menyiapkan Amazon SQS.

AWS SDK for Java 1.x

Untuk AWS SDK for Java 1.x, Anda dapat membuat yang AmazonSQSBufferedAsyncClient baru berdasarkan contoh berikut:

// Create the basic Amazon SQS async client
final AmazonSQSAsync sqsAsync = new AmazonSQSAsyncClient();
 
// Create the buffered client
final AmazonSQSAsync bufferedSqs = new AmazonSQSBufferedAsyncClient(sqsAsync);

Setelah Anda membuat yang baruAmazonSQSBufferedAsyncClient, Anda dapat menggunakannya untuk mengirim beberapa permintaan ke Amazon SQS (seperti yang Anda bisa denganAmazonSQSAsyncClient), misalnya:

final CreateQueueRequest createRequest = new CreateQueueRequest().withQueueName("MyQueue");
 
final CreateQueueResult res = bufferedSqs.createQueue(createRequest);
 
final SendMessageRequest request = new SendMessageRequest();
final String body = "Your message text" + System.currentTimeMillis();
request.setMessageBody( body );
request.setQueueUrl(res.getQueueUrl());
 
final Future<SendMessageResult> sendResult = bufferedSqs.sendMessageAsync(request);
 
final ReceiveMessageRequest receiveRq = new ReceiveMessageRequest()
    .withMaxNumberOfMessages(1)
    .withQueueUrl(queueUrl);
final ReceiveMessageResult rx = bufferedSqs.receiveMessage(receiveRq);

Mengkonfigurasi Amazon SQSBuffered AsyncClient

AmazonSQSBufferedAsyncClienttelah dikonfigurasi sebelumnya dengan pengaturan yang berfungsi untuk sebagian besar kasus penggunaan. Anda dapat mengkonfigurasi lebih lanjutAmazonSQSBufferedAsyncClient, misalnya:

1. Buat instance QueueBufferConfig kelas dengan parameter konfigurasi yang diperlukan.

2. Berikan instance ke AmazonSQSBufferedAsyncClient konstruktor.

// Create the basic Amazon SQS async client
final AmazonSQSAsync sqsAsync = new AmazonSQSAsyncClient();
 
final QueueBufferConfig config = new QueueBufferConfig()
    .withMaxInflightReceiveBatches(5)
    .withMaxDoneReceiveBatches(15);
 
// Create the buffered client
final AmazonSQSAsync bufferedSqs = new AmazonSQSBufferedAsyncClient(sqsAsync, config);

QueueBufferConfig parameter konfigurasi

Parameter	Nilai default	Deskripsi
longPoll	true	Ketika longPoll diatur ketrue, AmazonSQSBufferedAsyncClient mencoba untuk menggunakan polling panjang ketika mengkonsumsi pesan.
longPollWaitTimeoutSeconds	20 s	Jumlah waktu maksimum (dalam detik) yang diblokir ReceiveMessage panggilan di server, menunggu pesan muncul dalam antrian sebelum kembali dengan hasil penerimaan kosong. catatan Ketika polling panjang dinonaktifkan, pengaturan ini tidak berpengaruh.
maxBatchOpenMs	200 ms	Jumlah waktu maksimum (dalam milidetik) panggilan keluar menunggu panggilan lain yang dengannya ia mengumpulkan pesan dari jenis yang sama. Semakin tinggi pengaturan, semakin sedikit batch yang diperlukan untuk melakukan jumlah pekerjaan yang sama (namun, panggilan pertama dalam batch harus menghabiskan waktu lebih lama menunggu). Saat Anda menyetel parameter ini0, permintaan yang dikirimkan tidak menunggu permintaan lain, yang secara efektif menonaktifkan batching.
maxBatchSize	10 permintaan per batch	Jumlah maksimum pesan yang dikumpulkan bersama dalam satu permintaan. Semakin tinggi pengaturan, semakin sedikit batch yang diperlukan untuk melakukan jumlah permintaan yang sama. catatan 10 permintaan per batch adalah nilai maksimum yang diizinkan untuk Amazon SQS.
maxBatchSizeBytes	256 KiB	Ukuran maksimum kumpulan pesan, dalam byte, yang coba dikirim klien ke Amazon SQS. catatan 256 KiB adalah nilai maksimum yang diizinkan untuk Amazon SQS.
maxDoneReceiveBatches	10 batch	Jumlah maksimum batch penerima yang AmazonSQSBufferedAsyncClient mengambil dan menyimpan sisi klien. Semakin tinggi pengaturan, semakin banyak permintaan penerimaan yang dapat dipenuhi tanpa harus melakukan panggilan ke Amazon SQS (namun, semakin banyak pesan yang diambil sebelumnya, semakin lama mereka tetap berada di buffer, menyebabkan batas waktu visibilitas mereka sendiri kedaluwarsa). catatan 0menunjukkan bahwa semua pra-pengambilan pesan dinonaktifkan dan pesan hanya dikonsumsi sesuai permintaan.
maxInflightOutboundBatches	5 batch	Jumlah maksimum batch keluar aktif yang dapat diproses secara bersamaan. Semakin tinggi pengaturan, batch keluar yang lebih cepat dapat dikirim (tunduk pada kuota seperti CPU atau bandwidth) dan semakin banyak thread yang dikonsumsi oleh. AmazonSQSBufferedAsyncClient
maxInflightReceiveBatches	10 batch	Jumlah maksimum batch penerima aktif yang dapat diproses secara bersamaan. Semakin tinggi pengaturan, semakin banyak pesan yang dapat diterima (tunduk pada kuota seperti CPU atau bandwidth), dan semakin banyak utas yang dikonsumsi olehAmazonSQSBufferedAsyncClient. catatan 0menunjukkan bahwa semua pra-pengambilan pesan dinonaktifkan dan pesan hanya dikonsumsi sesuai permintaan.
visibilityTimeoutSeconds	-1	Ketika parameter ini disetel ke nilai positif, bukan nol, batas waktu visibilitas yang ditetapkan di sini akan mengganti batas waktu visibilitas yang ditetapkan pada antrian dari mana pesan dikonsumsi. catatan -1menunjukkan bahwa pengaturan default dipilih untuk antrian. Anda tidak dapat mengatur batas waktu visibilitas ke. 0

AWS SDK for Java 2.x

Untuk AWS SDK for Java 2.x, Anda dapat membuat yang SqsAsyncBatchManager baru berdasarkan contoh berikut:

// Create the basic Sqs Async Client
SqsAsyncClient sqs = SqsAsyncClient.builder() 
    .region(Region.US_EAST_1) 
    .build();

// Create the batch manager
SqsAsyncBatchManager sqsAsyncBatchManager = sqs.batchManager();

Setelah Anda membuat yang baruSqsAsyncBatchManager, Anda dapat menggunakannya untuk mengirim beberapa permintaan ke Amazon SQS (seperti yang Anda bisa denganSqsAsyncClient), misalnya:

final String queueName = "MyAsyncBufferedQueue" + UUID.randomUUID();
final CreateQueueRequest request = CreateQueueRequest.builder().queueName(queueName).build();
final String queueUrl = sqs.createQueue(request).join().queueUrl();
System.out.println("Queue created: " + queueUrl);


// Send messages
CompletableFuture<SendMessageResponse> sendMessageFuture;
for (int i = 0; i < 10; i++) {
    final int index = i;
    sendMessageFuture = sqsAsyncBatchManager.sendMessage(
            r -> r.messageBody("Message " + index).queueUrl(queueUrl));
    SendMessageResponse response= sendMessageFuture.join();
    System.out.println("Message " + response.messageId() + " sent!");
}

// Receive messages with customized configurations
CompletableFuture<ReceiveMessageResponse> receiveResponseFuture = customizedBatchManager.receiveMessage(
        r -> r.queueUrl(queueUrl)
                .waitTimeSeconds(10)
                .visibilityTimeout(20)
                .maxNumberOfMessages(10)
);
System.out.println("You have received " + receiveResponseFuture.join().messages().size() + " messages in total.");

// Delete messages
DeleteQueueRequest deleteQueueRequest =  DeleteQueueRequest.builder().queueUrl(queueUrl).build();
int code = sqs.deleteQueue(deleteQueueRequest).join().sdkHttpResponse().statusCode();
System.out.println("Queue is deleted, with statusCode " + code);

Mengonfigurasi SqsAsyncBatchManager

SqsAsyncBatchManagertelah dikonfigurasi sebelumnya dengan pengaturan yang berfungsi untuk sebagian besar kasus penggunaan. Anda dapat mengkonfigurasi lebih lanjutSqsAsyncBatchManager, misalnya:

Membuat konfigurasi khusus melaluiSqsAsyncBatchManager.Builder:

SqsAsyncBatchManager customizedBatchManager = SqsAsyncBatchManager.builder() 
    .client(sqs)
    .scheduledExecutor(Executors.newScheduledThreadPool(5))
    .overrideConfiguration(b -> b 
        .maxBatchSize(10)
        .sendRequestFrequency(Duration.ofMillis(200))
        .receiveMessageMinWaitDuration(Duration.ofSeconds(10))
        .receiveMessageVisibilityTimeout(Duration.ofSeconds(20)) 
        .receiveMessageAttributeNames(Collections.singletonList("*"))
        .receiveMessageSystemAttributeNames(Collections.singletonList(MessageSystemAttributeName.ALL)))
    .build();

Parameter BatchOverrideConfiguration

Parameter	Nilai default	Deskripsi
maxBatchSize	10 permintaan per batch	Jumlah maksimum pesan yang dikumpulkan bersama dalam satu permintaan. Semakin tinggi pengaturan, semakin sedikit batch yang diperlukan untuk melakukan jumlah permintaan yang sama. catatan Nilai maksimum yang diizinkan untuk Amazon SQS adalah 10 permintaan per batch.
sendRequestFrequency	200 ms	Jumlah waktu maksimum (dalam milidetik) panggilan keluar menunggu panggilan lain yang dengannya ia mengumpulkan pesan dari jenis yang sama. Semakin tinggi pengaturan, semakin sedikit batch yang diperlukan untuk melakukan jumlah pekerjaan yang sama (namun, panggilan pertama dalam batch harus menghabiskan waktu lebih lama menunggu). Saat Anda menyetel parameter ini0, permintaan yang dikirimkan tidak menunggu permintaan lain, yang secara efektif menonaktifkan batching.
receiveMessageVisibilityTimeout	-1	Ketika parameter ini disetel ke nilai positif, bukan nol, batas waktu visibilitas yang ditetapkan di sini akan mengganti batas waktu visibilitas yang ditetapkan pada antrian dari mana pesan dikonsumsi. catatan 1menunjukkan bahwa pengaturan default dipilih untuk antrian. Anda tidak dapat mengatur batas waktu visibilitas ke. 0
receiveMessageMinWaitDuration	50 ms	Jumlah waktu minimal (dalam milidetik) receiveMessage panggilan menunggu pesan yang tersedia diambil. Semakin tinggi pengaturan, semakin sedikit batch yang diperlukan untuk melakukan jumlah permintaan yang sama.