Reactor Kafka多topic并行消费是否需为每个topic单独创建KafkaReceiver

问题解答 #

1. poll操作是否是在日志打印request(NNN)...时触发？ #

不是直接对应触发关系。request(NNN)是Reactive Streams规范里的背压信号，代表下游已经空出了处理NNN条消息的容量。reactor-kafka内部会累计所有下游发来的请求数，只有当累计请求数达到Kafka消费者配置的fetch.min.bytes阈值，或者到了设置的poll超时时间，才会真正调用KafkaConsumer#poll拉取消息。打印request日志只是说明下游有了处理新消息的需求，不会每次打印都立刻执行poll操作。

2. 为什么有时是P-4、有时是r-coordinator-3打印onNext事件日志？ #

log()操作符的日志输出线程就是信号触发的线程：

• 你在分区流上加了publishOn(scheduler)后，该分区的后续消息处理会切换到你创建的P开头的并行调度器线程，所以这个分区的request信号会由P线程发起，当消息是已经拉取到本地、等待该分区处理的积压消息时，onNext信号就会在P线程上触发，日志就会显示P-4这类线程名。
• r-coordinator开头的线程是reactor-kafka内部原生Kafka消费者的事件调度线程，poll操作本身就是在这个线程上执行的。如果某条消息是刚poll返回、对应分区没有积压待处理的消息，消息会直接在r-coordinator线程上流转到log()节点，此时日志就会显示r-coordinator开头的线程名。

3. poll会不会拉取所有topic的消息，最终引发内存溢出？ #

你担心的问题确实存在。单个KafkaConsumer的poll操作会一次性拉取所有订阅Topic、所有分配分区的可用消息，拉取到的消息会先缓存在reactor-kafka的内部队列中。如果某个处理速度快的分区不断发送request信号，就会频繁触发poll操作，那些处理速度慢的分区的消息也会被一起拉取到内存中，积压到一定程度就会触发OOM。
你可以通过两个参数缓解这个问题：

• 调整Kafka消费者的max.poll.records参数，限制单次poll拉取的最大消息数
• 调整ReceiverOptions.maxInFlightPerPartition参数，限制每个分区最多允许多少条未确认的消息处于处理中
  但这两个参数只能降低风险，没法从根本上解决快分区触发poll拉取慢分区消息的问题。

4. 该场景下，我唯一的选择是为每个topic单独配置KafkaReceiver吗？ #

不是唯一选择，但分Topic创建独立的KafkaReceiver是最稳妥、维护成本最低的方案。不同Topic的消费完全隔离，处理速度慢的Topic不会影响其他Topic的消费，也不会出现跨Topic的消息内存积压问题。
如果不想创建多个KafkaReceiver，也可以选择两种替代方案：

• 按Topic维度分配独立的调度器和独立的背压控制逻辑
• 开启手动ACK机制，只有消息处理完成提交偏移量后，才发送新的request信号
  但这两种方案都需要额外开发逻辑，且还是不能完全避免快分区带动poll拉取慢分区消息的问题。如果你的不同Topic处理速度差异很大，优先推荐每个Topic单独创建KafkaReceiver，该方案额外消耗的资源极低，收益远高于维护复杂的单Receiver隔离逻辑。

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