Spring Cloud Stream中KStreams性能低下，如何通过并发优化？

我在Spring Boot中基于Spring Cloud Stream实现了Kafka Streams，从一个包含20个分区的主题读取数据，写入到分区数相同的主题，当前运行2个Pod，平均每分钟处理50万条记录。已设置并发数为10，但性能未得到提升，请问如何通过并发优化性能？

spring:
  cloud:
    config:
    function:
      definition: RiskProcessor1
    stream:
      bindings:
        RiskProcessor1-in-0:
          destination: ******
          consumer:
            concurrency: 10
      kafka:
        streams:
          binder:
            brokers: kaas-int.nam.nsroot.net:9093
            functions:
              RiskProcessor1:
                applicationId: RiskProcessor1_development
            configuration:
              commit.interval.ms: 1000
              security.protocol: SSL
              default:
                deserialization:
                  exception:
                    handler: org.apache.kafka.streams.errors.LogAndContinueExceptionHandler
              schema:
                registry:
                  url: *****:9081
          default:
            consumer:
              keySerde: ****
              valueSerde: ****
            producer:
              keySerde: ****
              valueSerde: ****

并发优化方案 #

1. 修正Kafka Streams线程数配置 #

你当前设置的consumer.concurrency是针对普通Spring Cloud Stream消费者的参数，对Kafka Streams绑定无效。Kafka Streams的并发由num.stream.threads控制，总线程数应匹配输入主题的分区数（20）。当前有2个Pod，每个Pod配置10个线程，刚好让每个线程处理一个分区，避免资源浪费或分配不均。

添加配置：

spring:
  cloud:
    stream:
      kafka:
        streams:
          binder:
            configuration:
              num.stream.threads: 10

2. 确保Pod有足够资源配额 #

每个Pod要支撑10个Streams线程，需分配足够的CPU和内存：

• CPU：每个线程至少0.5核，单Pod建议4-6核，避免上下文切换过载
• 内存：单Pod建议4-8GB，需覆盖线程运行、状态存储、序列化缓存等开销

3. 优化提交间隔与状态存储 #

• 调整commit.interval.ms：当前1000ms过于频繁，会增加磁盘IO和集群负担，建议改为5000-10000ms，减少提交次数提升吞吐量
• 切换到RocksDB状态存储：如果处理逻辑涉及聚合、窗口等状态操作，默认内存存储会受限，配置RocksDB并优化缓存参数：

spring:
  cloud:
    stream:
      kafka:
        streams:
          binder:
            configuration:
              state.dir: /tmp/kafka-streams-state
              rocksdb.config.setting:
                block.cache.size: 536870912  # 512MB缓存
                write.buffer.size: 134217728 # 128MB写缓冲区

4. 优化生产者批量发送 #

开启生产者批量发送和压缩，减少网络请求次数：

spring:
  cloud:
    stream:
      kafka:
        streams:
          default:
            producer:
              configuration:
                batch.size: 16384
                linger.ms: 5
                compression.type: snappy

5. 提升序列化效率 #

• 确认Schema Registry响应速度，可在Pod本地缓存Schema，减少远程调用开销
• 避免在Serde中执行复杂逻辑，将业务处理与序列化操作解耦

内容的提问来源于stack exchange，提问作者Himanshu Mahajan