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

# 使用 Kinesis
<a name="examples-kinesis"></a>

本節提供[Amazon Kinesis](https://docs.aws.amazon.com/kinesis/)使用 適用於 Java 的 AWS SDK 2.x 進行程式設計的範例。

如需 的詳細資訊 Kinesis，請參閱 [Amazon Kinesis 開發人員指南](https://docs.aws.amazon.com/streams/latest/dev/introduction.html)。

下列範例僅包含示範每個技術所需的程式碼。[GitHub 上提供完整程式碼範例](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javav2)。您可以從那裡下載單一原始檔案或將儲存庫複製到本機，以取得建置和執行的所有範例。

**Topics**
+ [訂閱 Amazon Kinesis Data Streams](examples-kinesis-stream.md)

# 訂閱 Amazon Kinesis Data Streams
<a name="examples-kinesis-stream"></a>

下列範例說明如何使用 `subscribeToShard`方法從 Amazon Kinesis 資料串流擷取和處理資料。 Kinesis Data Streams 現採用增強的廣發功能和低延遲 HTTP/2 資料擷取 API，讓開發人員更輕鬆地在相同的 Kinesis 資料串流上執行多個低延遲、高效能的應用程式。

## 設定
<a name="set-up"></a>

首先，建立非同步 Kinesis 用戶端和 [SubscribeToShardRequest](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/services/kinesis/model/SubscribeToShardRequest.html) 物件。這些物件用於下列每個範例，以訂閱 Kinesis 事件。

 **匯入** 

```
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicInteger;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import software.amazon.awssdk.core.async.SdkPublisher;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.kinesis.KinesisAsyncClient;
import software.amazon.awssdk.services.kinesis.model.ShardIteratorType;
import software.amazon.awssdk.services.kinesis.model.SubscribeToShardEvent;
import software.amazon.awssdk.services.kinesis.model.SubscribeToShardEventStream;
import software.amazon.awssdk.services.kinesis.model.SubscribeToShardRequest;
import software.amazon.awssdk.services.kinesis.model.SubscribeToShardResponse;
import software.amazon.awssdk.services.kinesis.model.SubscribeToShardResponseHandler;
```

 **Code** 

```
        Region region = Region.US_EAST_1;
        KinesisAsyncClient client = KinesisAsyncClient.builder()
        .region(region)
        .build();

        SubscribeToShardRequest request = SubscribeToShardRequest.builder()
                .consumerARN(CONSUMER_ARN)
                .shardId("arn:aws:kinesis:us-east-1:111122223333:stream/StockTradeStream")
                .startingPosition(s -> s.type(ShardIteratorType.LATEST)).build();
```

## 使用建置器界面
<a name="use-the-builder-interface"></a>

您可以使用 `builder` 方法，以簡化建立 [SubscribeToShardResponseHandler](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/services/kinesis/model/SubscribeToShardResponseHandler.html)。

使用 builder，您可以使用方法呼叫來設定每個生命週期回呼，而不用實作完整的介面。

 **Code** 

```
    private static CompletableFuture<Void> responseHandlerBuilder(KinesisAsyncClient client, SubscribeToShardRequest request) {
        SubscribeToShardResponseHandler responseHandler = SubscribeToShardResponseHandler
                .builder()
                .onError(t -> System.err.println("Error during stream - " + t.getMessage()))
                .onComplete(() -> System.out.println("All records stream successfully"))
                // Must supply some type of subscriber
                .subscriber(e -> System.out.println("Received event - " + e))
                .build();
        return client.subscribeToShard(request, responseHandler);
    }
```

如需進一步控制發佈者，您可以使用 `publisherTransformer` 方法來自訂發佈者。

 **Code** 

```
    private static CompletableFuture<Void> responseHandlerBuilderPublisherTransformer(KinesisAsyncClient client, SubscribeToShardRequest request) {
        SubscribeToShardResponseHandler responseHandler = SubscribeToShardResponseHandler
                .builder()
                .onError(t -> System.err.println("Error during stream - " + t.getMessage()))
                .publisherTransformer(p -> p.filter(e -> e instanceof SubscribeToShardEvent).limit(100))
                .subscriber(e -> System.out.println("Received event - " + e))
                .build();
        return client.subscribeToShard(request, responseHandler);
    }
```

請參閱 GitHub 上的[完整範例](https://github.com/awsdocs/aws-doc-sdk-examples/blob/ac748d8ef99cd17e297cb74fe13aa671e2679088/javav2/example_code/kinesis/src/main/java/com/example/kinesis/KinesisStreamEx.java)。

## 使用自訂回應處理常式
<a name="use-a-custom-response-handler"></a>

若要完全控制訂閱者和發佈者，請實作 `SubscribeToShardResponseHandler` 界面。

在這個範例中，您將實作 `onEventStream` 方法，允許您完整存取發佈者。此範例示範如何將發佈者轉換為事件記錄，供訂閱者列印。

 **Code** 

```
    private static CompletableFuture<Void> responseHandlerBuilderClassic(KinesisAsyncClient client, SubscribeToShardRequest request) {
        SubscribeToShardResponseHandler responseHandler = new SubscribeToShardResponseHandler() {

            @Override
            public void responseReceived(SubscribeToShardResponse response) {
                System.out.println("Receieved initial response");
            }

            @Override
            public void onEventStream(SdkPublisher<SubscribeToShardEventStream> publisher) {
                publisher
                        // Filter to only SubscribeToShardEvents
                        .filter(SubscribeToShardEvent.class)
                        // Flat map into a publisher of just records
                        .flatMapIterable(SubscribeToShardEvent::records)
                        // Limit to 1000 total records
                        .limit(1000)
                        // Batch records into lists of 25
                        .buffer(25)
                        // Print out each record batch
                        .subscribe(batch -> System.out.println("Record Batch - " + batch));
            }

            @Override
            public void complete() {
                System.out.println("All records stream successfully");
            }

            @Override
            public void exceptionOccurred(Throwable throwable) {
                System.err.println("Error during stream - " + throwable.getMessage());
            }
        };
        return client.subscribeToShard(request, responseHandler);
    }
```

請參閱 GitHub 上的[完整範例](https://github.com/awsdocs/aws-doc-sdk-examples/blob/ac748d8ef99cd17e297cb74fe13aa671e2679088/javav2/example_code/kinesis/src/main/java/com/example/kinesis/KinesisStreamEx.java)。

## 使用訪客界面
<a name="use-the-visitor-interface"></a>

您可以使用 [Visitor](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/services/kinesis/model/SubscribeToShardResponseHandler.Visitor.html) 物件，來訂閱您有興趣觀看的特定事件。

 **Code** 

```
    private static CompletableFuture<Void> responseHandlerBuilderVisitorBuilder(KinesisAsyncClient client, SubscribeToShardRequest request) {
        SubscribeToShardResponseHandler.Visitor visitor = SubscribeToShardResponseHandler.Visitor
                .builder()
                .onSubscribeToShardEvent(e -> System.out.println("Received subscribe to shard event " + e))
                .build();
        SubscribeToShardResponseHandler responseHandler = SubscribeToShardResponseHandler
                .builder()
                .onError(t -> System.err.println("Error during stream - " + t.getMessage()))
                .subscriber(visitor)
                .build();
        return client.subscribeToShard(request, responseHandler);
    }
```

請參閱 GitHub 上的[完整範例](https://github.com/awsdocs/aws-doc-sdk-examples/blob/ac748d8ef99cd17e297cb74fe13aa671e2679088/javav2/example_code/kinesis/src/main/java/com/example/kinesis/KinesisStreamEx.java)。

## 使用自訂訂閱者
<a name="use-a-custom-subscriber"></a>

您也可以實作您自己的自訂訂閱者，來訂閱串流。

此程式碼片段說明範例訂閱者。

 **Code** 

```
    private static class MySubscriber implements Subscriber<SubscribeToShardEventStream> {

        private Subscription subscription;
        private AtomicInteger eventCount = new AtomicInteger(0);

        @Override
        public void onSubscribe(Subscription subscription) {
            this.subscription = subscription;
            this.subscription.request(1);
        }

        @Override
        public void onNext(SubscribeToShardEventStream shardSubscriptionEventStream) {
            System.out.println("Received event " + shardSubscriptionEventStream);
            if (eventCount.incrementAndGet() >= 100) {
                // You can cancel the subscription at any time if you wish to stop receiving events.
                subscription.cancel();
            }
            subscription.request(1);
        }

        @Override
        public void onError(Throwable throwable) {
            System.err.println("Error occurred while stream - " + throwable.getMessage());
        }

        @Override
        public void onComplete() {
            System.out.println("Finished streaming all events");
        }
    }
```

您可以將自訂訂閱者傳遞至 `subscribe`方法，如下列程式碼片段所示。

 **Code** 

```
    private static CompletableFuture<Void> responseHandlerBuilderSubscriber(KinesisAsyncClient client, SubscribeToShardRequest request) {
        SubscribeToShardResponseHandler responseHandler = SubscribeToShardResponseHandler
                .builder()
                .onError(t -> System.err.println("Error during stream - " + t.getMessage()))
                .subscriber(MySubscriber::new)
                .build();
        return client.subscribeToShard(request, responseHandler);
    }
```

請參閱 GitHub 上的[完整範例](https://github.com/awsdocs/aws-doc-sdk-examples/blob/ac748d8ef99cd17e297cb74fe13aa671e2679088/javav2/example_code/kinesis/src/main/java/com/example/kinesis/KinesisStreamEx.java)。

## 將資料記錄寫入 Kinesis 資料串流
<a name="write-data-records-into-a-kinesis-data-stream"></a>

您可以使用 [KinesisClient](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/services/kinesis/KinesisClient.html) 物件，透過 `putRecords`方法將資料記錄寫入 Kinesis 資料串流。若要成功呼叫此方法，請建立 [PutRecordsRequest](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/services/kinesis/model/PutRecordsRequest.html) 物件。您可以將資料串流的名稱傳遞給 `streamName`方法。而且必須使用 `putRecords` 方法傳遞資料 (如下列程式碼範例所示)。

 **匯入** 

```
import software.amazon.awssdk.core.SdkBytes;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.kinesis.KinesisClient;
import software.amazon.awssdk.services.kinesis.model.PutRecordRequest;
import software.amazon.awssdk.services.kinesis.model.KinesisException;
import software.amazon.awssdk.services.kinesis.model.DescribeStreamRequest;
import software.amazon.awssdk.services.kinesis.model.DescribeStreamResponse;
```

在下列 Java 程式碼範例中，請注意 **StockTrade** 物件會用作寫入 Kinesis 資料串流的資料。在執行此範例之前，請確定您已建立資料串流。

 **Code** 

```
import software.amazon.awssdk.core.SdkBytes;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.kinesis.KinesisClient;
import software.amazon.awssdk.services.kinesis.model.PutRecordRequest;
import software.amazon.awssdk.services.kinesis.model.KinesisException;
import software.amazon.awssdk.services.kinesis.model.DescribeStreamRequest;
import software.amazon.awssdk.services.kinesis.model.DescribeStreamResponse;

/**
 * Before running this Java V2 code example, set up your development
 * environment, including your credentials.
 *
 * For more information, see the following documentation topic:
 *
 * https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/get-started.html
 */
public class StockTradesWriter {
    public static void main(String[] args) {
        final String usage = """

                Usage:
                    <streamName>

                Where:
                    streamName - The Amazon Kinesis data stream to which records are written (for example, StockTradeStream)
                """;

        if (args.length != 1) {
            System.out.println(usage);
            System.exit(1);
        }

        String streamName = args[0];
        Region region = Region.US_EAST_1;
        KinesisClient kinesisClient = KinesisClient.builder()
                .region(region)
                .build();

        // Ensure that the Kinesis Stream is valid.
        validateStream(kinesisClient, streamName);
        setStockData(kinesisClient, streamName);
        kinesisClient.close();
    }

    public static void setStockData(KinesisClient kinesisClient, String streamName) {
        try {
            // Repeatedly send stock trades with a 100 milliseconds wait in between.
            StockTradeGenerator stockTradeGenerator = new StockTradeGenerator();

            // Put in 50 Records for this example.
            int index = 50;
            for (int x = 0; x < index; x++) {
                StockTrade trade = stockTradeGenerator.getRandomTrade();
                sendStockTrade(trade, kinesisClient, streamName);
                Thread.sleep(100);
            }

        } catch (KinesisException | InterruptedException e) {
            System.err.println(e.getMessage());
            System.exit(1);
        }
        System.out.println("Done");
    }

    private static void sendStockTrade(StockTrade trade, KinesisClient kinesisClient,
            String streamName) {
        byte[] bytes = trade.toJsonAsBytes();

        // The bytes could be null if there is an issue with the JSON serialization by
        // the Jackson JSON library.
        if (bytes == null) {
            System.out.println("Could not get JSON bytes for stock trade");
            return;
        }

        System.out.println("Putting trade: " + trade);
        PutRecordRequest request = PutRecordRequest.builder()
                .partitionKey(trade.getTickerSymbol()) // We use the ticker symbol as the partition key, explained in
                                                       // the Supplemental Information section below.
                .streamName(streamName)
                .data(SdkBytes.fromByteArray(bytes))
                .build();

        try {
            kinesisClient.putRecord(request);
        } catch (KinesisException e) {
            System.err.println(e.getMessage());
        }
    }

    private static void validateStream(KinesisClient kinesisClient, String streamName) {
        try {
            DescribeStreamRequest describeStreamRequest = DescribeStreamRequest.builder()
                    .streamName(streamName)
                    .build();

            DescribeStreamResponse describeStreamResponse = kinesisClient.describeStream(describeStreamRequest);

            if (!describeStreamResponse.streamDescription().streamStatus().toString().equals("ACTIVE")) {
                System.err.println("Stream " + streamName + " is not active. Please wait a few moments and try again.");
                System.exit(1);
            }

        } catch (KinesisException e) {
            System.err.println("Error found while describing the stream " + streamName);
            System.err.println(e);
            System.exit(1);
        }
    }
}
```

請參閱 GitHub 上的[完整範例](https://github.com/awsdocs/aws-doc-sdk-examples/blob/master/javav2/example_code/kinesis/src/main/java/com/example/kinesis/StockTradesWriter.java)。

## 使用第三方程式庫
<a name="use-a-third-party-library"></a>

您可以使用其他第三方程式庫，而非實作自訂的訂閱者。此範例示範使用 RxJava 實作，但您可以使用實作反應式串流介面的任何程式庫。如需該程式庫的詳細資訊，請參閱 [Github 上的 RxJava wiki 頁面](https://github.com/ReactiveX/RxJava/wiki)。

若要使用該程式庫，請將其新增做為相依性。如果您使用 Maven，範例會顯示要使用的 POM 片段。

 **POM 項目** 

```
<dependency>
 <groupId>io.reactivex.rxjava2</groupId>
 <artifactId>rxjava</artifactId>
 <version>2.2.21</version>
</dependency>
```

 **匯入** 

```
import java.net.URI;
import java.util.concurrent.CompletableFuture;

import io.reactivex.Flowable;
import software.amazon.awssdk.auth.credentials.ProfileCredentialsProvider;
import software.amazon.awssdk.core.async.SdkPublisher;
import software.amazon.awssdk.http.Protocol;
import software.amazon.awssdk.http.SdkHttpConfigurationOption;
import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.kinesis.KinesisAsyncClient;
import software.amazon.awssdk.services.kinesis.model.ShardIteratorType;
import software.amazon.awssdk.services.kinesis.model.StartingPosition;
import software.amazon.awssdk.services.kinesis.model.SubscribeToShardEvent;
import software.amazon.awssdk.services.kinesis.model.SubscribeToShardRequest;
import software.amazon.awssdk.services.kinesis.model.SubscribeToShardResponseHandler;
import software.amazon.awssdk.utils.AttributeMap;
```

此範例使用 `onEventStream` 生命週期方法中的 RxJava。這讓您能夠完整存取發佈者，後者可用於建立 Rx Flowable。

 **Code** 

```
        SubscribeToShardResponseHandler responseHandler = SubscribeToShardResponseHandler
            .builder()
            .onError(t -> System.err.println("Error during stream - " + t.getMessage()))
            .onEventStream(p -> Flowable.fromPublisher(p)
                                        .ofType(SubscribeToShardEvent.class)
                                        .flatMapIterable(SubscribeToShardEvent::records)
                                        .limit(1000)
                                        .buffer(25)
                                        .subscribe(e -> System.out.println("Record batch = " + e)))
            .build();
```

您也可以使用 `publisherTransformer` 方法搭配 `Flowable` 發佈者。您必須將 `Flowable` 發佈者調整為 *SdkPublisher*，如以下範例所示。

 **Code** 

```
        SubscribeToShardResponseHandler responseHandler = SubscribeToShardResponseHandler
            .builder()
            .onError(t -> System.err.println("Error during stream - " + t.getMessage()))
            .publisherTransformer(p -> SdkPublisher.adapt(Flowable.fromPublisher(p).limit(100)))
            .build();
```

請參閱 GitHub 上的[完整範例](https://github.com/awsdocs/aws-doc-sdk-examples/blob/main/javav2/example_code/kinesis/src/main/java/com/example/kinesis/KinesisStreamRxJavaEx.java)。

## 其他資訊
<a name="more-information"></a>
+  Amazon Kinesis API 參考中的 [SubscribeToShardEvent](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_SubscribeToShardEvent.html) 
+  Amazon Kinesis API 參考中的 [SubscribeToShard](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_SubscribeToShard.html)