我正在运行一个kafka流应用程序，它有三个子拓扑。活动的阶段大致如下： stream 主题a selectKey 和重新划分主题 to 专题b stream 专题b foreach 专题b至专题c
Producer stream 主题c
主题c to 主题d
主题a、b和c都是具体化的，这意味着如果每个主题有40个分区，我的最大并行度是120。
起初，我运行5个流应用程序，每个流应用程序有8个线程。在这种情况下，我遇到了不一致的表现。似乎有些共享同一线程的子拓扑比其他子拓扑更需要cpu，过了一段时间，我会遇到以下错误： Member [client_id] in group [consumer_group] has failed, removing it from the group (kafka.coordinator.group.GroupCoordinator) . 一切都将得到重新平衡，这可能导致性能下降，直到下一次失败和重新平衡。
我的问题如下：
多个子拓扑如何能够在单个线程上运行？投票队列？
每个线程如何决定如何将计算资源分配给它的每个子拓扑？
在这种情况下，如何优化线程与主题的分区比率以避免周期性的使用者失败？e、 例如，1:1的比例能保证更稳定的性能吗？
如果使用1:1的比率，如何确保每个线程都分配了自己的主题分区，并且一些线程没有空闲？
1条答案
按热度按时间anhgbhbe1#
线程将轮询（）不同子拓扑的所有主题并检查记录
topic元数据以将其提供给正确的任务。每个子拓扑的处理方式都是相同的，也就是说，如果您愿意，可用资源是均匀分布的。
只有拥有足够的内核时，1:1的比例才有用。我建议监视您的cpu利用率。如果太高（大于80%），您应该添加更多的核心/线程。
Kafka流自动为您处理。
一些一般性意见：
你可以考虑增加
max.poll.interval.ms配置以避免使用者退出组你可以考虑减少
max.poll.records以减少每个poll()从而减少两次连续呼叫之间的时间poll().请注意
max.poll.records并不意味着增加网络/代理通信—如果单个获取请求返回的记录多于max.poll.records配置时，数据只是缓冲在使用者和下一个poll()将从缓冲数据提供服务，从而避免代理往返