Kafka Streams KTable接收输入流新消息未即时更新的优化咨询

• 首先测试代码中的KafkaProducer.send()是异步API，调用后不会等待消息写入Kafka Broker就直接返回，你新增的xyz消息可能还没完成发送，Streams程序自然消费不到。可以在send调用后加.get()阻塞等待发送结果，或者调用sandboxProducer.flush()强制将缓冲区消息全部刷到Broker。
• Kafka Streams默认对状态存储做了批量更新优化，两个核心配置导致了更新延迟：
  1. commit.interval.ms：至少一次语义下默认值为30000ms（30秒），Exactly Once语义下默认值为100ms，该参数控制Streams多久将内存中的状态变更持久化到本地存储、并提交消费位点。
  2. cache.max.bytes.buffering：默认值为10MB，Streams会将状态更新暂存在内存缓存中，要么缓存占满、要么到达提交间隔才会将变更刷到磁盘状态存储，未刷入的缓存在查询状态存储时不可见。

配置调整 #

在你的streamsConfiguration中添加以下配置即可大幅降低延迟：

// 调小提交间隔，根据可接受的延迟调整，这里示例为100ms
streamsConfiguration.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100);
// 如果需要更新立刻可见，完全禁用状态缓存，缓存大小设为0
streamsConfiguration.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0);
// 调优消费端拉取配置，降低消息拉取等待时间
streamsConfiguration.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 100);
streamsConfiguration.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1);

生产场景权衡 #

如果不是必须要求毫秒级更新可见，建议不要完全禁用缓存，保留部分缓存可以合并同一个key的多次更新，大幅降低磁盘写入量和changelog Topic的写入流量，提升批量处理性能。比如可以接受1秒延迟的话，将commit.interval.ms设为1000即可兼顾延迟和性能，完全适配你每小时300MB的批量写入场景。

测试代码调整 #

调整配置后，在发送完新消息和第二次打印之间加一个1~2秒的等待，给Streams预留消费处理的时间，即可看到最新的更新结果。

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