本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

# Amazon SQS 批次動作
<a name="sqs-batch-api-actions"></a>

Amazon SQS 提供批次動作，可協助您降低成本，並使用單一動作操作多達 10 則訊息。這些批次動作包括：
+ `[SendMessageBatch](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_SendMessageBatch.html)`
+ `[DeleteMessageBatch](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_DeleteMessageBatch.html)`
+ `[ChangeMessageVisibilityBatch](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_ChangeMessageVisibilityBatch.html)`

使用批次動作，您可以在單一 API 呼叫中執行多個操作，這有助於最佳化效能並降低成本。您可以使用查詢 API 或任何支援 Amazon SQS 批次動作的 AWS SDK 來利用批次功能。

**重要詳細資訊**
+ **訊息大小限制：**單一`SendMessageBatch`呼叫中傳送的所有訊息總大小不得超過 1，048，576 位元組 (1 MiB)
+ **許可：**您無法明確設定 `SendMessageBatch`、 `DeleteMessageBatch`或 的許可`ChangeMessageVisibilityBatch`。反之，為 `SendMessage`、 `DeleteMessage`或 設定許可，為對應的 動作批次版本`ChangeMessageVisibility`設定許可。
+ **主控台支援：**Amazon SQS 主控台不支援批次動作。您必須使用查詢 API 或 AWS SDK 來執行批次操作。

## 批次訊息動作
<a name="batching-message-actions"></a>

若要進一步最佳化成本和效率，請考慮下列批次處理訊息動作的最佳實務：
+ **批次 API 動作：**使用 [Amazon SQS 批次 API 動作](#sqs-batch-api-actions)來傳送、接收和刪除訊息，以及使用單一動作變更多個訊息的訊息可見性逾時。這可減少 API 呼叫的數量和相關聯的成本。
+ **用戶端緩衝和長輪詢：**使用長輪詢搭配 隨附的[緩衝非同步用戶端](sqs-client-side-buffering-request-batching.md)，將用戶端緩衝與請求批次結合在一起 適用於 Java 的 AWS SDK。此方法有助於將請求數量降至最低，並最佳化大量訊息的處理。

**注意**  
Amazon SQS 緩衝非同步用戶端目前不支援 FIFO 佇列。

# 使用 Amazon SQS 啟用用戶端緩衝和請求批次處理
<a name="sqs-client-side-buffering-request-batching"></a>

[適用於 Java 的 AWS SDK](https://aws.amazon.com/sdkforjava/) 包含可存取 Amazon SQS 的 `AmazonSQSBufferedAsyncClient`。此用戶端允許使用用戶端緩衝，進行簡單的請求批次處理。從用戶端發出的呼叫會先緩衝，然後以批次請求的形式傳送至 Amazon SQS。

用戶端緩衝功能最多可緩衝處理 10 個請求並以批次請求的形式傳送，降低了使用 Amazon SQS 的成本且能減少傳送的請求數。`AmazonSQSBufferedAsyncClient` 會對同步和非同步的呼叫進行緩衝處理。批次處理的請求以及對[長輪詢](sqs-short-and-long-polling.md)的支援也有助於提高傳輸量。如需詳細資訊，請參閱 [透過 Amazon SQS 使用水平擴展和動作批次來增加輸送量](sqs-throughput-horizontal-scaling-and-batching.md)。

由於 `AmazonSQSBufferedAsyncClient` 實作的介面與 `AmazonSQSAsyncClient` 相同，從 `AmazonSQSAsyncClient` 移轉到 `AmazonSQSBufferedAsyncClient` 通常只需對既有的程式碼進行最少的更動。

**注意**  
Amazon SQS 緩衝非同步用戶端目前不支援 FIFO 佇列。

## 使用 AmazonSQSBufferedAsyncClient
<a name="using-buffered-async-client"></a>

開始之前，請完成 [設定 Amazon SQS](sqs-setting-up.md) 中的步驟。

### AWS 適用於 Java 的 SDK 1.x
<a name="using-buffered-async-client-java1"></a>

對於適用於 Java 的 AWS SDK 1.x，您可以`AmazonSQSBufferedAsyncClient`根據下列範例建立新的 ：

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

新的 `AmazonSQSBufferedAsyncClient` 建立完成後，您可以使用它來傳送多個請求至 Amazon SQS (如同使用 `AmazonSQSAsyncClient`)，例如：

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

### 設定 AmazonSQSBufferedAsyncClient
<a name="configuring-buffered-async-client"></a>

`AmazonSQSBufferedAsyncClient` 已預先設為適合大多數使用案例的組態。您可以進一步設定 `AmazonSQSBufferedAsyncClient`，例如：

1. 使用必要的組態參數，建立 `QueueBufferConfig` 類別的執行個體。

1. 對 `AmazonSQSBufferedAsyncClient` 建構函式提供此執行個體。

```
// 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 組態參數**  

| 參數 | 預設值 | Description | 
| --- | --- | --- | 
| longPoll | true |  若 `longPoll` 設為 `true`，`AmazonSQSBufferedAsyncClient` 在消費訊息時會嘗試使用長輪詢。  | 
| longPollWaitTimeoutSeconds | 20 秒 |  在傳回空的接收結果前，`ReceiveMessage` 呼叫留置於伺服器上等待訊息出現在佇列中的時間上限 (單位為秒)。  停用長輪詢時，此設定沒有作用。   | 
| maxBatchOpenMs | 200 毫秒 |  傳出呼叫對於其他也要批次處理同類型訊息的呼叫稍作等待的時間上限 (單位為毫秒)。 此設定值越高，執行相同工作量所需的批次數就越少 (但是，批次的第一次呼叫就需要花越長時間等待)。 若將此參數設為 `0`，提交的請求便不會等待其他請求，實際上即是停用了批次處理功能。  | 
| maxBatchSize | 每批次 10 個請求 |  單次請求同時批次處理的訊息數上限。此設定值越高，執行相同請求數所需的批次數量就越少。  Amazon SQS 允許的上限值為每批次 10 個請求。   | 
| maxBatchSizeBytes | 1 MiB |  用戶端嘗試傳送至 Amazon SQS 之訊息批次的大小上限，單位為位元組。  1 MiB 是 Amazon SQS 允許的最大值。   | 
| maxDoneReceiveBatches | 10 個批次 |  `AmazonSQSBufferedAsyncClient` 在用戶端預取並存放的接收批次數上限。 此設定值越高，則不必呼叫 Amazon SQS 就能滿足越多的接收請求 (但是，預取的訊息越多，停留在緩衝區內的時間就越久，而導致訊息的可見性逾時會到期)。  `0` 表示已停用所有訊息預先擷取，且訊息只會隨需使用。   | 
| maxInflightOutboundBatches | 5 個批次 |  可同時處理的活動中傳出批次的上限。 此設定值越高，傳出批次的傳送速度越快 (受限於 CPU 或頻寬等的配額)，且 `AmazonSQSBufferedAsyncClient` 所耗用的執行緒越多。  | 
| maxInflightReceiveBatches | 10 個批次 |  可同時處理的活動中接收批次的上限。 此設定值越高，可接收的訊息數越多 (受限於 CPU 或頻寬等的配額)，且 `AmazonSQSBufferedAsyncClient` 所耗用的執行緒越多。  `0` 表示已停用所有訊息預先擷取，且訊息只會隨需使用。   | 
| visibilityTimeoutSeconds | -1 |  若將此參數設為非零正值，其設定的可見性逾時值就會覆寫消費訊息的佇列所設的可見性逾時值。  `-1` 表示將選擇佇列預設的設定。 可見性逾時不可設為 `0`。   | 

### AWS 適用於 Java 的 SDK 2.x
<a name="using-buffered-async-client-java2"></a>

對於適用於 Java 的 AWS SDK 2.x，您可以`SqsAsyncBatchManager`根據下列範例建立新的 ：

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

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

新的 `SqsAsyncBatchManager` 建立完成後，您可以使用它來傳送多個請求至 Amazon SQS (如同使用 `SqsAsyncClient`)，例如：

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

### 設定 SqsAsyncBatchManager
<a name="configuring-SqsAsyncBatchManager"></a>

`SqsAsyncBatchManager` 已預先設為適合大多數使用案例的組態。您可以進一步設定 `SqsAsyncBatchManager`，例如：

透過 建立自訂組態`SqsAsyncBatchManager.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();
```


**`BatchOverrideConfiguration` 參數**  

| 參數 | 預設值 | Description | 
| --- | --- | --- | 
| maxBatchSize |  每批次 10 個請求  | 單次請求同時批次處理的訊息數上限。此設定值越高，執行相同請求數所需的批次數量就越少。  Amazon SQS 允許的值上限為每個批次 10 個請求。  | 
| sendRequestFrequency |  200 毫秒  | 傳出呼叫對於其他也要批次處理同類型訊息的呼叫稍作等待的時間上限 (單位為毫秒)。 此設定值越高，執行相同工作量所需的批次數就越少 (但是，批次的第一次呼叫就需要花越長時間等待)。 若將此參數設為 `0`，提交的請求便不會等待其他請求，實際上即是停用了批次處理功能。 | 
| receiveMessageVisibilityTimeout |  -1  | 若將此參數設為非零正值，其設定的可見性逾時值就會覆寫消費訊息的佇列所設的可見性逾時值。   `1` 表示將選擇佇列預設的設定。可見性逾時不可設為 `0`。   | 
| receiveMessageMinWaitDuration |  50 毫秒  | `receiveMessage` 呼叫等待擷取可用訊息的時間下限 （以毫秒為單位）。設定越高，執行相同數量的請求所需的批次就越少。  | 

# 透過 Amazon SQS 使用水平擴展和動作批次來增加輸送量
<a name="sqs-throughput-horizontal-scaling-and-batching"></a>

Amazon SQS 支援高輸送量傳訊。如需輸送量限制的詳細資訊，請參閱 [Amazon SQS 訊息配額](quotas-messages.md)。

若要最大化輸送量：
+ 透過新增更多每個 的執行個體，水平[擴展](#horizontal-scaling)生產者和消費者。
+ 使用[動作批次](#request-batching)在單一請求中傳送或接收多則訊息，減少 API 呼叫額外負荷。

## 水平擴展
<a name="horizontal-scaling"></a>

由於您是透過 HTTP 請求-回應協定來存取 Amazon SQS，*請求延遲* (自開始請求至收到回應的時間間隔) 會限制您透過單一連線使用單一執行緒所能達到的輸送量。例如，若 Amazon EC2 型的用戶端到 Amazon SQS 的延遲度在相同區域的平均值為 20 毫秒，透過單一連線使用單一執行緒可達到的最大輸送量平均為 50 TPS。

*水平擴展*意指增加訊息生產者 (提出 `[SendMessage](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_SendMessage.html)` 請求) 及消費者 (提出 `[ReceiveMessage](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_ReceiveMessage.html)` 和 `[DeleteMessage](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_DeleteMessage.html)` 請求) 的數目，藉以提高佇列整體的傳輸量。水平擴展的方式有三種：
+ 增加每個用戶端的執行緒數目
+ 增加更多用戶端
+ 增加每個用戶端的執行緒數目並增加更多用戶端

增加更多用戶端之後，基本上佇列傳輸量會獲得線性增益。例如，若用戶端數目加倍，傳輸量也會翻倍。

## 動作批次處理
<a name="request-batching"></a>

*批次處理*可在每次往返服務期間處理更多工作 (例如，透過單次 `SendMessageBatch` 請求傳送多則訊息)。Amazon SQS 批次動作包括 `[SendMessageBatch](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_SendMessageBatch.html)`、`[DeleteMessageBatch](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_DeleteMessageBatch.html)` 和 `[ChangeMessageVisibilityBatch](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_ChangeMessageVisibilityBatch.html)`。若要在不變動生產者或消費者的情況下利用批次處理功能，您可以使用 [Amazon SQS 緩衝非同步用戶端](sqs-client-side-buffering-request-batching.md)。

**注意**  
由於 `[ReceiveMessage](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_ReceiveMessage.html)` 一次可處理 10 則訊息，也就不必有 `ReceiveMessageBatch` 動作。

批次處理會將批次動作的延遲度分散到批次請求中的多則訊息，而非接受單則訊息 (例如 `[SendMessage](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_SendMessage.html)` 請求) 的所有延遲。由於每次往返所執行的工作較多，批次請求可以有效率地運用執行緒和連線，進而提高傳輸量。

您可以結合批次處理與水平擴展來提供傳輸量，所需的執行緒、連線和請求的數目會比單獨的訊息請求所需的量更少。您可以使用批次的 Amazon SQS 動作一次傳送、接收、刪除多達 10 則訊息。由於 Amazon SQS 收費是以請求數為準，批次處理可大幅降低您的成本。

批次處理可能會給您的應用程式帶來一些複雜性 (例如，您的應用程式必須將訊息累積起來後再傳送出去，或者有時候必須花更久的時間等待回應)。但是，批次處理在以下情況仍然頗具效益：
+ 應用程式會在短時間內產生許多訊息，所以絕不會延遲太久。
+ 訊息消費者自行從佇列擷取訊息，有別於典型的訊息生產者需要傳送訊息來回應其無法控制的事件。

**重要**  
即使批次中的個別訊息失敗，批次請求也可能會成功。提出批次請求之後，務必確認個別訊息是否出現失敗情況，必要時重試動作。

## 單次操作和批次請求實際可行的 Java 範例
<a name="working-java-example-batch-requests"></a>

### 先決條件
<a name="batch-request-java-example-prerequisites"></a>

新增 `aws-java-sdk-sqs.jar`、`aws-java-sdk-ec2.jar`，以及 `commons-logging.jar` 套件到您的 Java 建置類別路徑。以下範例顯示 Maven 專案 `pom.xml` 檔案中的此等依存關係。

```
<dependencies>
    <dependency>
        <groupId>com.amazonaws</groupId>
        <artifactId>aws-java-sdk-sqs</artifactId>
        <version>LATEST</version>
    </dependency>
    <dependency>
        <groupId>com.amazonaws</groupId>
        <artifactId>aws-java-sdk-ec2</artifactId>
        <version>LATEST</version>
    </dependency>
    <dependency>
        <groupId>commons-logging</groupId>
        <artifactId>commons-logging</artifactId>
        <version>LATEST</version>
    </dependency>
</dependencies>
```

### SimpleProducerConsumer.java
<a name="batch-request-java-example-code"></a>

以下 Java 程式碼範例實作單純的生產者-消費者模式。主要的執行緒會產生數個生產者和消費者執行緒，負責在指定時間處理 1 KB 的訊息。此範例包括提出單次操作請求的生產者和消費者，以及提出批次請求的生產者和消費者。

```
/*
 * Copyright 2010-2024 Amazon.com, Inc. or its affiliates. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License").
 * You may not use this file except in compliance with the License.
 * A copy of the License is located at
 *
 *  https://aws.amazon.com/apache2.0
 *
 * or in the "license" file accompanying this file. This file is distributed
 * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
 * express or implied. See the License for the specific language governing
 * permissions and limitations under the License.
 *
 */

import com.amazonaws.AmazonClientException;
import com.amazonaws.ClientConfiguration;
import com.amazonaws.services.sqs.AmazonSQS;
import com.amazonaws.services.sqs.AmazonSQSClientBuilder;
import com.amazonaws.services.sqs.model.*;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.math.BigInteger;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.Scanner;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Start a specified number of producer and consumer threads, and produce-consume
 * for the least of the specified duration and 1 hour. Some messages can be left
 * in the queue because producers and consumers might not be in exact balance.
 */
public class SimpleProducerConsumer {

    // The maximum runtime of the program.
    private final static int MAX_RUNTIME_MINUTES = 60;
    private final static Log log = LogFactory.getLog(SimpleProducerConsumer.class);

    public static void main(String[] args) throws InterruptedException {

        final Scanner input = new Scanner(System.in);

        System.out.print("Enter the queue name: ");
        final String queueName = input.nextLine();

        System.out.print("Enter the number of producers: ");
        final int producerCount = input.nextInt();

        System.out.print("Enter the number of consumers: ");
        final int consumerCount = input.nextInt();

        System.out.print("Enter the number of messages per batch: ");
        final int batchSize = input.nextInt();

        System.out.print("Enter the message size in bytes: ");
        final int messageSizeByte = input.nextInt();

        System.out.print("Enter the run time in minutes: ");
        final int runTimeMinutes = input.nextInt();

        /*
         * Create a new instance of the builder with all defaults (credentials
         * and region) set automatically. For more information, see Creating
         * Service Clients in the AWS SDK for Java Developer Guide.
         */
        final ClientConfiguration clientConfiguration = new ClientConfiguration()
                .withMaxConnections(producerCount + consumerCount);

        final AmazonSQS sqsClient = AmazonSQSClientBuilder.standard()
                .withClientConfiguration(clientConfiguration)
                .build();

        final String queueUrl = sqsClient
                .getQueueUrl(new GetQueueUrlRequest(queueName)).getQueueUrl();

        // The flag used to stop producer, consumer, and monitor threads.
        final AtomicBoolean stop = new AtomicBoolean(false);

        // Start the producers.
        final AtomicInteger producedCount = new AtomicInteger();
        final Thread[] producers = new Thread[producerCount];
        for (int i = 0; i < producerCount; i++) {
            if (batchSize == 1) {
                producers[i] = new Producer(sqsClient, queueUrl, messageSizeByte,
                        producedCount, stop);
            } else {
                producers[i] = new BatchProducer(sqsClient, queueUrl, batchSize,
                        messageSizeByte, producedCount,
                        stop);
            }
            producers[i].start();
        }

        // Start the consumers.
        final AtomicInteger consumedCount = new AtomicInteger();
        final Thread[] consumers = new Thread[consumerCount];
        for (int i = 0; i < consumerCount; i++) {
            if (batchSize == 1) {
                consumers[i] = new Consumer(sqsClient, queueUrl, consumedCount,
                        stop);
            } else {
                consumers[i] = new BatchConsumer(sqsClient, queueUrl, batchSize,
                        consumedCount, stop);
            }
            consumers[i].start();
        }

        // Start the monitor thread.
        final Thread monitor = new Monitor(producedCount, consumedCount, stop);
        monitor.start();

        // Wait for the specified amount of time then stop.
        Thread.sleep(TimeUnit.MINUTES.toMillis(Math.min(runTimeMinutes,
                MAX_RUNTIME_MINUTES)));
        stop.set(true);

        // Join all threads.
        for (int i = 0; i < producerCount; i++) {
            producers[i].join();
        }

        for (int i = 0; i < consumerCount; i++) {
            consumers[i].join();
        }

        monitor.interrupt();
        monitor.join();
    }

    private static String makeRandomString(int sizeByte) {
        final byte[] bs = new byte[(int) Math.ceil(sizeByte * 5 / 8)];
        new Random().nextBytes(bs);
        bs[0] = (byte) ((bs[0] | 64) & 127);
        return new BigInteger(bs).toString(32);
    }

    /**
     * The producer thread uses {@code SendMessage}
     * to send messages until it is stopped.
     */
    private static class Producer extends Thread {
        final AmazonSQS sqsClient;
        final String queueUrl;
        final AtomicInteger producedCount;
        final AtomicBoolean stop;
        final String theMessage;

        Producer(AmazonSQS sqsQueueBuffer, String queueUrl, int messageSizeByte,
                 AtomicInteger producedCount, AtomicBoolean stop) {
            this.sqsClient = sqsQueueBuffer;
            this.queueUrl = queueUrl;
            this.producedCount = producedCount;
            this.stop = stop;
            this.theMessage = makeRandomString(messageSizeByte);
        }

        /*
         * The producedCount object tracks the number of messages produced by
         * all producer threads. If there is an error, the program exits the
         * run() method.
         */
        public void run() {
            try {
                while (!stop.get()) {
                    sqsClient.sendMessage(new SendMessageRequest(queueUrl,
                            theMessage));
                    producedCount.incrementAndGet();
                }
            } catch (AmazonClientException e) {
                /*
                 * By default, AmazonSQSClient retries calls 3 times before
                 * failing. If this unlikely condition occurs, stop.
                 */
                log.error("Producer: " + e.getMessage());
                System.exit(1);
            }
        }
    }

    /**
     * The producer thread uses {@code SendMessageBatch}
     * to send messages until it is stopped.
     */
    private static class BatchProducer extends Thread {
        final AmazonSQS sqsClient;
        final String queueUrl;
        final int batchSize;
        final AtomicInteger producedCount;
        final AtomicBoolean stop;
        final String theMessage;

        BatchProducer(AmazonSQS sqsQueueBuffer, String queueUrl, int batchSize,
                      int messageSizeByte, AtomicInteger producedCount,
                      AtomicBoolean stop) {
            this.sqsClient = sqsQueueBuffer;
            this.queueUrl = queueUrl;
            this.batchSize = batchSize;
            this.producedCount = producedCount;
            this.stop = stop;
            this.theMessage = makeRandomString(messageSizeByte);
        }

        public void run() {
            try {
                while (!stop.get()) {
                    final SendMessageBatchRequest batchRequest =
                            new SendMessageBatchRequest().withQueueUrl(queueUrl);

                    final List<SendMessageBatchRequestEntry> entries =
                            new ArrayList<SendMessageBatchRequestEntry>();
                    for (int i = 0; i < batchSize; i++)
                        entries.add(new SendMessageBatchRequestEntry()
                                .withId(Integer.toString(i))
                                .withMessageBody(theMessage));
                    batchRequest.setEntries(entries);

                    final SendMessageBatchResult batchResult =
                            sqsClient.sendMessageBatch(batchRequest);
                    producedCount.addAndGet(batchResult.getSuccessful().size());

                    /*
                     * Because SendMessageBatch can return successfully, but
                     * individual batch items fail, retry the failed batch items.
                     */
                    if (!batchResult.getFailed().isEmpty()) {
                        log.warn("Producer: retrying sending "
                                + batchResult.getFailed().size() + " messages");
                        for (int i = 0, n = batchResult.getFailed().size();
                             i < n; i++) {
                            sqsClient.sendMessage(new
                                    SendMessageRequest(queueUrl, theMessage));
                            producedCount.incrementAndGet();
                        }
                    }
                }
            } catch (AmazonClientException e) {
                /*
                 * By default, AmazonSQSClient retries calls 3 times before
                 * failing. If this unlikely condition occurs, stop.
                 */
                log.error("BatchProducer: " + e.getMessage());
                System.exit(1);
            }
        }
    }

    /**
     * The consumer thread uses {@code ReceiveMessage} and {@code DeleteMessage}
     * to consume messages until it is stopped.
     */
    private static class Consumer extends Thread {
        final AmazonSQS sqsClient;
        final String queueUrl;
        final AtomicInteger consumedCount;
        final AtomicBoolean stop;

        Consumer(AmazonSQS sqsClient, String queueUrl, AtomicInteger consumedCount,
                 AtomicBoolean stop) {
            this.sqsClient = sqsClient;
            this.queueUrl = queueUrl;
            this.consumedCount = consumedCount;
            this.stop = stop;
        }

        /*
         * Each consumer thread receives and deletes messages until the main
         * thread stops the consumer thread. The consumedCount object tracks the
         * number of messages that are consumed by all consumer threads, and the
         * count is logged periodically.
         */
        public void run() {
            try {
                while (!stop.get()) {
                    try {
                        final ReceiveMessageResult result = sqsClient
                                .receiveMessage(new
                                        ReceiveMessageRequest(queueUrl));

                        if (!result.getMessages().isEmpty()) {
                            final Message m = result.getMessages().get(0);
                            sqsClient.deleteMessage(new
                                    DeleteMessageRequest(queueUrl,
                                    m.getReceiptHandle()));
                            consumedCount.incrementAndGet();
                        }
                    } catch (AmazonClientException e) {
                        log.error(e.getMessage());
                    }
                }
            } catch (AmazonClientException e) {
                /*
                 * By default, AmazonSQSClient retries calls 3 times before
                 * failing. If this unlikely condition occurs, stop.
                 */
                log.error("Consumer: " + e.getMessage());
                System.exit(1);
            }
        }
    }

    /**
     * The consumer thread uses {@code ReceiveMessage} and {@code
     * DeleteMessageBatch} to consume messages until it is stopped.
     */
    private static class BatchConsumer extends Thread {
        final AmazonSQS sqsClient;
        final String queueUrl;
        final int batchSize;
        final AtomicInteger consumedCount;
        final AtomicBoolean stop;

        BatchConsumer(AmazonSQS sqsClient, String queueUrl, int batchSize,
                      AtomicInteger consumedCount, AtomicBoolean stop) {
            this.sqsClient = sqsClient;
            this.queueUrl = queueUrl;
            this.batchSize = batchSize;
            this.consumedCount = consumedCount;
            this.stop = stop;
        }

        public void run() {
            try {
                while (!stop.get()) {
                    final ReceiveMessageResult result = sqsClient
                            .receiveMessage(new ReceiveMessageRequest(queueUrl)
                                    .withMaxNumberOfMessages(batchSize));

                    if (!result.getMessages().isEmpty()) {
                        final List<Message> messages = result.getMessages();
                        final DeleteMessageBatchRequest batchRequest =
                                new DeleteMessageBatchRequest()
                                        .withQueueUrl(queueUrl);

                        final List<DeleteMessageBatchRequestEntry> entries =
                                new ArrayList<DeleteMessageBatchRequestEntry>();
                        for (int i = 0, n = messages.size(); i < n; i++)
                            entries.add(new DeleteMessageBatchRequestEntry()
                                    .withId(Integer.toString(i))
                                    .withReceiptHandle(messages.get(i)
                                            .getReceiptHandle()));
                        batchRequest.setEntries(entries);

                        final DeleteMessageBatchResult batchResult = sqsClient
                                .deleteMessageBatch(batchRequest);
                        consumedCount.addAndGet(batchResult.getSuccessful().size());

                        /*
                         * Because DeleteMessageBatch can return successfully,
                         * but individual batch items fail, retry the failed
                         * batch items.
                         */
                        if (!batchResult.getFailed().isEmpty()) {
                            final int n = batchResult.getFailed().size();
                            log.warn("Producer: retrying deleting " + n
                                    + " messages");
                            for (BatchResultErrorEntry e : batchResult
                                    .getFailed()) {

                                sqsClient.deleteMessage(
                                        new DeleteMessageRequest(queueUrl,
                                                messages.get(Integer
                                                        .parseInt(e.getId()))
                                                        .getReceiptHandle()));

                                consumedCount.incrementAndGet();
                            }
                        }
                    }
                }
            } catch (AmazonClientException e) {
                /*
                 * By default, AmazonSQSClient retries calls 3 times before
                 * failing. If this unlikely condition occurs, stop.
                 */
                log.error("BatchConsumer: " + e.getMessage());
                System.exit(1);
            }
        }
    }

    /**
     * This thread prints every second the number of messages produced and
     * consumed so far.
     */
    private static class Monitor extends Thread {
        private final AtomicInteger producedCount;
        private final AtomicInteger consumedCount;
        private final AtomicBoolean stop;

        Monitor(AtomicInteger producedCount, AtomicInteger consumedCount,
                AtomicBoolean stop) {
            this.producedCount = producedCount;
            this.consumedCount = consumedCount;
            this.stop = stop;
        }

        public void run() {
            try {
                while (!stop.get()) {
                    Thread.sleep(1000);
                    log.info("produced messages = " + producedCount.get()
                            + ", consumed messages = " + consumedCount.get());
                }
            } catch (InterruptedException e) {
                // Allow the thread to exit.
            }
        }
    }
}
```

### 監控範本執行階段的數量指標
<a name="batch-request-java-example-monitoring-metrics"></a>

Amazon SQS 會自動為傳送、接收及刪除的訊息產生數量指標。您可以透過佇列的**監控**索引標籤或 [CloudWatch 主控台](https://console.aws.amazon.com/cloudwatch/home)存取這類指標和其他項目。

**注意**  
可能要在佇列開始執行達 15 分鐘後方可取得指標。