如何在基于 kafka 流的应用程序中降低滞后

Question

我有一个真正的env与3个kafka机器集群，它正在接收大量数据。对于每个主题，有 25 个分区，复制因子设置为 2。

我的应用程序（基于kafka流的应用程序）从这个kafka集群获取数据的应用程序停机了一个多月。现在，每个分区都有大量滞后；高达90000000。

我知道以下参数：

max.poll.records ; default —> 500
max.partition.fetch.bytes ; default —> 1048576
fetch.max.bytes ; default —> 52428800
fetch.min.bytes ; default —> 1

max.poll.interval.ms ; default —> 300000
request.timeout.ms; default —> 30000
session.timeout.ms ; default —> 10000

我有两个消费者节点(使用kafka集群数据的相同组id)。

然而，它并没有赶上滞后，而是保持不变。有人能建议如何改进它以降低滞后吗？

Answer 1

如果您的应用程序关闭了一个月，则某些记录已过期，因为主题中的默认保留期为 7 天，因此您很可能丢失了一些消息。此外，默认偏移重置的保留期为 1 天或 7 天，具体取决于您的 Kafka 流版本。似乎你有自动.offset.reset：最早，所以它开始在每个分区的开头消费消息。如果需要跳过所有消息而仅使用新消息，则应设置 auto.offset.reset： 最新，application.id 值更改为新值。

如果您希望并行使用消息并加快延迟的减少，您可以将config < code > num . stream . threads 设置为类似于< code>12的值(< code > num . stream . threads * number of consumer nodes 应该小于或等于< code>numberOfPartitions，否则，一些线程将处于空闲状态)，或者需要增加使用者节点的数量。