消费者之间的话题分享答案

【问题标题】：Topic sharing between consumers消费者之间的话题分享
【发布时间】：2020-09-08 19:47:28
【问题描述】：

我正在开发一个 Springboot 应用程序，它会为 Kafka 主题中的不同分区重新平衡（例如，700 个主题，每个主题有 10 个分区，即 7000 个分区）。但是我想启动多个 dockerized 应用程序实例，其中应用程序将包含所有 700 个主题名称，但它应该只选择前 25 个分区并取消订阅其他分区）

@KafkaListener( topics = "#{kafkaProperties.getTopics()}" )

kafkaProperties.getTopics() 返回所有 700 个主题名称

【问题讨论】：

这个不清楚；当每个主题只有 10 个时，“前 25 个分区”是什么意思？更详细地描述您的要求。如果您的意思是您希望每个实例仅获得 25 个分区，例如c1 - t1(10), t2(10), t3(5)、c2 - t3(5), t4(10), t5(10) 等，您将需要自定义 ConsumerPartitionAssignor。
@GaryRussell 是的，这正是我所需要的，但现在发生的是每个主题分区有一个线程。我参考了这个post 来为每个主题分区创建一个线程，但现在我想将分区分布在多个正在运行的实例中，如果一个实例由于任何原因出现故障，其他实例应该选择负载，直到新实例出现增益.
@GaryRussell 让我进一步澄清一下，将运行应用程序的多个实例，例如。多个 jars 或 dockerized 应用程序实例每个都应该具有相同的配置，即所有主题都已配置，但每个实例应该选择有限数量的线程，每个线程都有一个主题/分区，并让其他实例继续使用。现在第二个实例应该从平衡主题中挑选一些，以此类推。

标签： spring-boot apache-kafka spring-kafka

【解决方案1】：

正如我在最初要求澄清的评论中所说，对于这样的分区分配方案，您需要实现自定义 ConsumerPartitionAssignor。

每个实例应选择有限数量的线程，每个线程具有一个主题/分区

实例不会“挑选”主题/分区；选择一个实例，它决定哪个实例获取哪个主题/分区。

查看它的 javadocs。

/**
 * This interface is used to define custom partition assignment for use in
 * {@link org.apache.kafka.clients.consumer.KafkaConsumer}. Members of the consumer group subscribe
 * to the topics they are interested in and forward their subscriptions to a Kafka broker serving
 * as the group coordinator. The coordinator selects one member to perform the group assignment and
 * propagates the subscriptions of all members to it. Then {@link #assign(Cluster, GroupSubscription)} is called
 * to perform the assignment and the results are forwarded back to each respective members
 *
 * In some cases, it is useful to forward additional metadata to the assignor in order to make
 * assignment decisions. For this, you can override {@link #subscriptionUserData(Set)} and provide custom
 * userData in the returned Subscription. For example, to have a rack-aware assignor, an implementation
 * can use this user data to forward the rackId belonging to each member.
 */
public interface ConsumerPartitionAssignor {

要向实例添加更多线程（消费者），请增加容器并发。

对于大量实例，我建议使用COOPERATIVE 重新平衡协议。

【讨论】：