Spring Cloud Stream：处理消息需要很长时间时对StreamListener的正确处理

Question

当 StreamListener 需要很长时间（比max.poll.interval.ms）处理消息时，该特定消费者被占用，其他新消息将被分配到其他分区。时间大于max.poll.interval.ms后，再平衡发生，同样的情况也会发生在另一个消费者身上。因此，此消息将在所有分区中循环并继续占用资源。

但是，这种情况并不经常发生，只有几条消息不知何故需要这么长时间来处理，而且是无法控制的。

我们可以提交偏移量并在几次重新平衡后将其扔给 DLQ 吗？如果是，我们该怎么做？如果不是，这种情况的正确处理是什么？

Answer 1

增加max.poll.interval.ms 不会对性能产生影响（除非检测到真的死亡的消费者需要更长的时间）。

每次处理此“不良”记录时进行重新平衡对性能的损害更大。

但是，您可以使用自定义 SeekToCurrentErrorHandler together with a recoverer such as the DeadLetterPublishingRecoverer 做您想做的事情。您还需要一个重新平衡侦听器来计算重新平衡和一些机制来跨实例共享来自错误处理程序的状态（标准只将状态保存在内存中）。

我认为相当复杂。