使用Spring Cloud Stream库重试将消息推送到Kinesis

Question

我正在编写一个生产者，使用Spring云流库将消息推送到Kinesis流。我能够成功地将数据推送到kinesis，但在kinesis方面，它失败了，吞吐量超出异常。有没有办法再次重试推送这些消息并确切知道哪个消息失败了？此外，我不想使用KPL或KCL。

我尝试了答案中建议的解决方案，这是我的配置：

Spring云流动绑定。输入制作人errorChannelEnabled：true spring。云流动绑定。输入制作人错误目的地：myFooDestination。myGroup。错误

这是正确的方法吗？那么，如何将“spring.cloud.stream.bindings.input.producer.error.destination:myFooDestination.myGroup.errors”映射到spring集成配置中的“error channel”属性，如下所示？

<int-http:inbound-channel-adapter id="abcErrorChannel"
                                      channel="defChannel"
                                      **error-channel="errorChannel**"
                                  </int-http:inbound-channel-adapter>

Answer 1

客户组仅适用于消费者。在生产者方面没有这样的组概念。对于在生产者方面绑定到错误通道，请使用下面的属性和下面的代码绑定到错误通道。

spring.cloud.stream.bindings.output.destination=kinesis-stream
spring.cloud.stream.bindings.output.producer.errorChannelEnabled=true

@ServiceActivator(inputChannel = "kinesis-stream.errors")
public void receiveProduceError(Message receiveMsg) {
    System.err.println("receive error msg: " + receiveMsg);
}

Answer 2

AWS Kinesis活页夹中的PutRecord请求完全基于AmazonKinesisAsync，发送到AWS确实是异步的。因此，我们不能在那里使用内置的RetryTemplate功能。但同时，该异步操作的错误被发送到特定于目标的错误通道：https://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/#_producer_properties

ErrorChannel已启用

When set to true, if the binder supports asynchroous send results, send failures are sent to an error channel for the destination. See “[binder-error-channels]” for more information.

默认值：false。

频道名称基于目的地和消费者组，加上后缀：https://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/#spring-云流概述错误处理

此外，如果您绑定到现有目标，例如：

spring.cloud.stream.bindings.input.destination=myFooDestination
spring.cloud.stream.bindings.input.group=myGroup

完整的目标名称是myFODestination.myGroup，然后专用错误通道名称是myFODestination.myGroup.errors。