Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

# Travaillez avec Kinesis
<a name="examples-kinesis"></a>

Cette section fournit des exemples de programmation à l'[Amazon Kinesis](https://docs.aws.amazon.com/kinesis/)aide de la version AWS SDK pour Java 2.x.

Pour plus d'informations Kinesis, consultez le [guide du Amazon Kinesis développeur](https://docs.aws.amazon.com/streams/latest/dev/introduction.html).

Les exemples suivants incluent uniquement le code nécessaire pour démontrer chaque technique. L'[exemple de code complet est disponible sur GitHub](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javav2). À partir de là, vous pouvez télécharger un fichier source unique ou cloner le référentiel en local pour obtenir tous les exemples à générer et exécuter.

**Topics**
+ [Abonnez-vous à Amazon Kinesis Data Streams](examples-kinesis-stream.md)

# Abonnez-vous à Amazon Kinesis Data Streams
<a name="examples-kinesis-stream"></a>

Les exemples suivants vous montrent comment récupérer et traiter des données à partir de flux de Amazon Kinesis données à l'aide de `subscribeToShard` cette méthode. Kinesis Data Streams utilise désormais la fonction fanout améliorée et une API de récupération de données HTTP/2 à faible latence, ce qui permet aux développeurs d'exécuter plus facilement plusieurs applications à faible latence et hautes performances sur le même flux de données. Kinesis 

## Configuration
<a name="set-up"></a>

Créez d'abord un Kinesis client asynchrone et un [SubscribeToShardRequest](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/services/kinesis/model/SubscribeToShardRequest.html)objet. Ces objets sont utilisés dans chacun des exemples suivants pour s'abonner à Kinesis des événements.

 **Importations** 

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

## Utiliser l'interface du générateur
<a name="use-the-builder-interface"></a>

Vous pouvez utiliser `builder` cette méthode pour simplifier la création du [SubscribeToShardResponseHandler](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/services/kinesis/model/SubscribeToShardResponseHandler.html).

À l'aide du générateur, vous pouvez définir chaque rappel de cycle de vie avec un appel de méthode au lieu d'implémenter l'intégralité de l'interface.

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

Pour plus de contrôle sur l'éditeur, vous pouvez utiliser la méthode `publisherTransformer` pour personnaliser l'éditeur.

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

Consultez l'[exemple complet](https://github.com/awsdocs/aws-doc-sdk-examples/blob/ac748d8ef99cd17e297cb74fe13aa671e2679088/javav2/example_code/kinesis/src/main/java/com/example/kinesis/KinesisStreamEx.java) sur GitHub.

## Utiliser un gestionnaire de réponses personnalisé
<a name="use-a-custom-response-handler"></a>

Pour un contrôle total de l'abonné et de l'éditeur, implémentez l'`SubscribeToShardResponseHandler`interface.

Dans cet exemple, vous implémentez la méthode `onEventStream` qui autorise un accès complet à l'éditeur. Cet exemple montre comment transformer l'éditeur en enregistrements d'événement pour impression par l'abonné.

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

Consultez l'[exemple complet](https://github.com/awsdocs/aws-doc-sdk-examples/blob/ac748d8ef99cd17e297cb74fe13aa671e2679088/javav2/example_code/kinesis/src/main/java/com/example/kinesis/KinesisStreamEx.java) sur GitHub.

## Utiliser l'interface visiteur
<a name="use-the-visitor-interface"></a>

Vous pouvez utiliser un objet [Visitor](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/services/kinesis/model/SubscribeToShardResponseHandler.Visitor.html) pour vous abonner à des événements spécifiques que vous souhaitez surveiller.

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

Consultez l'[exemple complet](https://github.com/awsdocs/aws-doc-sdk-examples/blob/ac748d8ef99cd17e297cb74fe13aa671e2679088/javav2/example_code/kinesis/src/main/java/com/example/kinesis/KinesisStreamEx.java) sur GitHub.

## Utiliser un abonné personnalisé
<a name="use-a-custom-subscriber"></a>

Vous pouvez également implémenter votre propre abonné personnalisé pour s'abonner au flux.

Cet extrait de code montre un exemple d'abonné.

 **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");
        }
    }
```

Vous pouvez transmettre l'abonné personnalisé à la `subscribe` méthode comme indiqué dans l'extrait de code suivant.

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

Consultez l'[exemple complet](https://github.com/awsdocs/aws-doc-sdk-examples/blob/ac748d8ef99cd17e297cb74fe13aa671e2679088/javav2/example_code/kinesis/src/main/java/com/example/kinesis/KinesisStreamEx.java) sur GitHub.

## Écrire des enregistrements de données dans un flux Kinesis de données
<a name="write-data-records-into-a-kinesis-data-stream"></a>

Vous pouvez utiliser l'[KinesisClient](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/services/kinesis/KinesisClient.html)objet pour écrire des enregistrements de données dans un flux de Kinesis données à l'aide de la `putRecords` méthode. Pour appeler correctement cette méthode, créez un [PutRecordsRequest](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/services/kinesis/model/PutRecordsRequest.html)objet. Vous transmettez le nom du flux de données à la `streamName` méthode. Vous devez également transmettre les données en utilisant la méthode `putRecords` (comme indiqué dans l'exemple de code suivant).

 **Importations** 

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

Dans l'exemple de code Java suivant, notez que **StockTrade**l'objet est utilisé comme donnée pour écrire dans le flux de Kinesis données. Avant d'exécuter cet exemple, assurez-vous d'avoir créé le flux de données.

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

Consultez l'[exemple complet](https://github.com/awsdocs/aws-doc-sdk-examples/blob/master/javav2/example_code/kinesis/src/main/java/com/example/kinesis/StockTradesWriter.java) sur GitHub.

## Utiliser une bibliothèque tierce
<a name="use-a-third-party-library"></a>

Vous pouvez utiliser d'autres bibliothèques tierces au lieu d'implémenter un abonné personnalisé. Cet exemple montre comment utiliser l' RxJava implémentation, mais vous pouvez utiliser n'importe quelle bibliothèque qui implémente les interfaces Reactive Streams. Consultez la [page RxJava wiki sur Github](https://github.com/ReactiveX/RxJava/wiki) pour plus d'informations sur cette bibliothèque.

Pour utiliser la bibliothèque, ajoutez-la en tant que dépendance. Si vous utilisez Maven, l'exemple illustre l'extrait POM à utiliser.

 **Entrée POM** 

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

 **Importations** 

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

Cet exemple est utilisé RxJava dans la méthode du `onEventStream` cycle de vie. Cet exemple vous donne un accès complet à l'éditeur, qui peut être utilisé pour créer un 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();
```

Vous pouvez également utiliser la méthode `publisherTransformer` avec l'éditeur `Flowable`. Vous devez adapter l'`Flowable`éditeur à un *SdkPublisher*, comme indiqué dans l'exemple suivant.

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

Consultez l'[exemple complet](https://github.com/awsdocs/aws-doc-sdk-examples/blob/main/javav2/example_code/kinesis/src/main/java/com/example/kinesis/KinesisStreamRxJavaEx.java) sur GitHub.

## En savoir plus
<a name="more-information"></a>
+  [SubscribeToShardEvent](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_SubscribeToShardEvent.html)dans la référence de Amazon Kinesis l'API
+  [SubscribeToShard](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_SubscribeToShard.html)dans la référence de Amazon Kinesis l'API