Весенняя облачная струя производителя, партия

Question

Я занимаюсь разработкой проекта с использованием версии 1.0.0 механизма связывания кинесиса облачного потока (https://github.com/spring-cloud/spring-cloud-stream-binder-aws-kinesis). У меня есть входящий канал (поток кинезиса 1), который используется моим приложением. По сути, мое приложение читаетизменение и запись нового сообщения в исходящий канал (поток kinesis 2).

Я могу читать сообщения из 100 200 записей одновременно, но когда дело доходит до записи в исходящий поток, я не могуПосмотрите, как настроить исходящий канал для выполнения пакетной записи, эквивалентной PutRecordsRequest API потока Kinesis.

Кто-нибудь мог выполнить что-то подобное с выходными каналами облачного потока?

Спасибо.

Answer 1

Вы можете просто вернуть PutRecordsRequest из вашего @StreamListener и настроить .producer.useNativeEncoding = true для цели привязки вывода.Таким образом, результат @StreamListener не будет преобразован в byte[], и это станет обязанностью KinesisMessageHandler для правильной обработки этой PutRecordsRequest полезной нагрузки.И это теперь возможно:

if (message.getPayload() instanceof PutRecordsRequest) {
        AsyncHandler<PutRecordsRequest, PutRecordsResult> asyncHandler =
                obtainAsyncHandler(message, (PutRecordsRequest) message.getPayload());

        return this.amazonKinesis.putRecordsAsync((PutRecordsRequest) message.getPayload(), asyncHandler);
    }