水平缩放spring kafka消费者应用程序

Question

我想知道什么是相对于最大水平扩展实例数配置分区数量的好方法。

假设我有一个有6个分区的主题。

我有一个应用程序，它使用的ConnettKafkaListenerContainerFactory与setConcurrency的6.这意味着我将有6个KafkaMessageListenerContainer，每个都使用一个线程，并且均匀地消耗来自所有分区的消息。

如果以上是正确的，那么我想知道如果我通过添加另一个实例水平缩放应用程序会发生什么？如果新实例具有相同的配置，并发为6，当然也具有相同的消费者组，我相信第二个实例不会消耗任何消息。因为不会发生重新平衡，因为每个现有的消费者将有一个分区分配给它。

但是，如果我们回到第一个示例，有6个分区，其中一个实例的并发性为3，那么每个使用者线程/KafkaMessageListenerContainer将分配2个分区。如果我们扩展这个应用程序（相同的用户组id和3的并发性），我相信会发生重新平衡，两个实例将分别从3个分区进行消费。

这些假设是否正确，如果不正确，你应该如何处理这种情况？

Answer 1

一般来说，对于默认行为，您的假设是正确的，默认行为基于：

/**
 * <p>The range assignor works on a per-topic basis. For each topic, we lay out the available partitions in numeric order
 * and the consumers in lexicographic order. We then divide the number of partitions by the total number of
 * consumers to determine the number of partitions to assign to each consumer. If it does not evenly
 * divide, then the first few consumers will have one extra partition.
 *
 * <p>For example, suppose there are two consumers <code>C0</code> and <code>C1</code>, two topics <code>t0</code> and
 * <code>t1</code>, and each topic has 3 partitions, resulting in partitions <code>t0p0</code>, <code>t0p1</code>,
 * <code>t0p2</code>, <code>t1p0</code>, <code>t1p1</code>, and <code>t1p2</code>.
 *
 * <p>The assignment will be:
 * <ul>
 * <li><code>C0: [t0p0, t0p1, t1p0, t1p1]</code></li>
 * <li><code>C1: [t0p2, t1p2]</code></li>
 * </ul>
 *
 * Since the introduction of static membership, we could leverage <code>group.instance.id</code> to make the assignment behavior more sticky.
 * For the above example, after one rolling bounce, group coordinator will attempt to assign new <code>member.id</code> towards consumers,
 * for example <code>C0</code> -&gt; <code>C3</code> <code>C1</code> -&gt; <code>C2</code>.
 *
 * <p>The assignment could be completely shuffled to:
 * <ul>
 * <li><code>C3 (was C0): [t0p2, t1p2] (before was [t0p0, t0p1, t1p0, t1p1])</code>
 * <li><code>C2 (was C1): [t0p0, t0p1, t1p0, t1p1] (before was [t0p2, t1p2])</code>
 * </ul>
 *
 * The assignment change was caused by the change of <code>member.id</code> relative order, and
 * can be avoided by setting the group.instance.id.
 * Consumers will have individual instance ids <code>I1</code>, <code>I2</code>. As long as
 * 1. Number of members remain the same across generation
 * 2. Static members' identities persist across generation
 * 3. Subscription pattern doesn't change for any member
 *
 * <p>The assignment will always be:
 * <ul>
 * <li><code>I0: [t0p0, t0p1, t1p0, t1p1]</code>
 * <li><code>I1: [t0p2, t1p2]</code>
 * </ul>
 */
public class RangeAssignor extends AbstractPartitionAssignor {

但是，您可以通过分区插入任何消费者分区赋值器。分配策略消费者财产：https://kafka.apache.org/documentation/#consumerconfigs_partition.assignment.strategy

另请参见ConsumerPartitionAssignorJavaDocs，了解更多信息及其实现，以便为您的用例做出选择。