0

我的应用程序从一个流中消费，然后将消息推送到三个流上

粘合剂：

public interface MyBinder {

  @Input("input1")
  SubscribableChannel input1();

  @Output("output1")
  MessageChannel output1();

  @Output("output2")
  MessageChannel output2();

  @Output("output3")
  MessageChannel output3();


}

配置：

spring:
  cloud:
    stream:
      kinesis:
        binder:
          locks:
            leaseDuration: 30
            refreshPeriod: 3000
        bindings:
          input1:
            consumer:
              listenerMode: batch
              recordsLimit: 1500
              idleBetweenPolls: 10000
              consumer-backoff: 1000
      bindings:
        input1:
          group: my-group
          destination: input1-stream
          content-type: application/json
        output1:
          destination: output1-stream
          content-type: application/json
        output2:
          destination: output2-stream
          content-type: application/json
        output3:
          destination: output3-stream
          content-type: application/json

我们推送的数据流在每条记录中大约 800kb。我们看到 AbstractAwsMessageHandler/AmazonKinesisAsyncClient 中有更多数据导致非常频繁的 GC 刷新。

我们使用的是 1.0.0.RELEASE 版本的 Binder

你能帮忙吗？

4

1 回答 1

2

只有我可以通过您的配置说您将拥有1500 * 3 PutRecordRequest实例，AbstractAwsMessageHandler并且由于默认情况下处于async模式，您可能会导致等待 AWS 服务处理它们的队列开销。

您可以考虑减少recordsLimit或将所有生产者配置为处于sync模式：https ://github.com/spring-cloud/spring-cloud-stream-binder-aws-kinesis/blob/master/spring-cloud-stream -binder-kinesis-docs/src/main/asciidoc/overview.adoc#kinesis-producer-properties

如果要消耗的记录较少，您将在内存中拥有较少的对象。在同步生产模式的情况下，您将阻塞消费者线程，因此它不会从输入流中提取更多记录。

于 2019-04-17T13:49:05.293 回答