Aws Kinesis потребляет реактивное сообщение - PullRequest
Новая
       115

Aws Kinesis потребляет реактивное сообщение

2 голосов
/ 07 августа 2020

Я пытаюсь реализовать реактивную реакцию потребителя кинезиса. На данный момент я понимаю, что осколок кинезиса всегда используется потоком, поэтому поток, который его использует, подписывается.

Вот мой код: 

   private static CompletableFuture<Void> responseHandlerBuilderReactor(KinesisAsyncClient client, SubscribeToShardRequest request) {

        SubscribeToShardResponseHandler responseHandler = SubscribeToShardResponseHandler
                .builder()
                .onError(t -> System.err.println("Error during stream - " + t.getMessage()))
                .onEventStream(p -> Flux.from(p)
                        .ofType(SubscribeToShardEvent.class)
                        .flatMapIterable(SubscribeToShardEvent::records)
                        .limitRate(2)
                        .buffer(2)
                        .subscribe(e -> {
                            // decoder.decode(record.data()).toString();
                            e.forEach(el-> {
                                try {
                                    System.out.println(decoder.decode(el.data().asByteBuffer()));
                                } catch (CharacterCodingException e1) {
                                    e1.printStackTrace();
                                }
                            });
                        }))
                        .build();
        return client.subscribeToShard(request, responseHandler);

    }

    public static void main(String[] args) {

        KinesisAsyncClient client = KinesisAsyncClient.create();

        SubscribeToShardRequest request = SubscribeToShardRequest.builder()
                .consumerARN(CONSUMER_ARN)
                .shardId("shardId-000000000000")
                .startingPosition(StartingPosition.builder().type(ShardIteratorType.TRIM_HORIZON).build())
                .build();

        responseHandlerBuilderReactor(client, request).join();

        client.close();
    }

Я думаю, что есть много деталей, которые я не знаю Я не понимаю, что такое кинезис, и даже если я продолжу читать, он не станет более ясным.

В документации говорится, что это соединение будет иметь соединение HTTP / 2 с низкой задержкой, но я понятия не имею, как справиться с этим между работает, что я имею в виду, если я перезапущу приложение, я не хочу обрабатывать уже использованные сообщения. Как я могу реализовать RecordProcessorCheckpointer для реактивной версии?

Также через несколько минут процесс завершится без ошибок. Я помню, как читал в документации, что мне нужно повторно подписаться. Достаточно, если я бегу в бесконечном l oop: responseHandlerBuilderReactor(client, request).join();?

...