Kafka 消费者重新平衡

2025年5月14日 | 11 分钟阅读

Apache Kafka 是一个高度可伸缩、容错、分布式的流处理平台，专为实时数据处理而设计。其核心是主题（topics）、分区（partitions）、生产者（producers）、消费者（consumers）和代理（brokers）。主题是消息的流，每个主题都被分区以实现消费者之间的并行处理。Kafka 消费者从主题读取记录，通常多个消费者构成一个消费者组 (consumer group)，这有助于分发读取主题分区的负载。这就是消费者重组算法 (consumer rebalancing algorithm) 发挥作用的地方。

消费者重组 (Consumer rebalancing) 是指当消费者组的组成发生变化时（例如，消费者加入或离开组），重新分配消费者组内分区给各个消费者的过程。这确保了工作负载能够高效地分布在所有可用的消费者之间。如果没有适当的重组机制，一些消费者可能会不堪重负，而另一些则处于空闲状态。

Kafka 中何时会发生重组

Kafka 中的重组是一个基本过程，它确保了消费者组中消费者之间的分区能够最优地分配。这个过程对于维护高可用性、负载均衡和容错至关重要。了解重组何时发生对于 Kafka 用户优化他们的应用程序和最大限度地减少中断至关重要。

触发重组的关键场景

重组会在几种情况下触发

当一个新的消费者加入 Kafka 中的某个消费者组时，系统会经历一个称为重组的过程。这个过程有助于在组内所有消费者之间平均分配分区，从而使每个消费者都有公平的工作量。

示例

假设你有一个主题，最初有 6 个分区，并且有 2 个消费者正在从中消费

消费者 1：分区 0、1、2

消费者 2：分区 3、4、5

当新的消费者 3 加入时，分区将被重新分配

消费者 1：分区 0、1

消费者 2：分区 2、3

消费者 3：分区 4、5

消费者离开组：当一个消费者崩溃、停止或以其他方式离开消费者组时，Kafka 会启动一次重组，将已离开消费者的分区重新分配给剩余的消费者。此操作可确保消息处理不会中断。

示例

使用之前的场景，如果消费者 2 离开了组，它处理的分区需要被重新分配

消费者 1：分区 0、1、2

消费者 3：分区 3、4、5

主题分区更改：如果向主题添加了新的分区，则会触发重组，将这些新分区分配给组内的消费者。此更改允许更好的扩展和工作负载分布。

示例

如果主题最初有 6 个分区，并且分区数量增加到 8 个，则需要重新分配现有消费者

消费者 1：分区 0、1、2、3

消费者 2：分区 4、5

消费者 3：分区 6、7

配置更改：消费者组配置的更改，例如修改会话超时或组 ID，也可能触发重组。这通常不太常见，但在动态环境中可能会发生。

手动重组：开发者可以通过调用离开和重新加入消费者组的方法来触发手动重组。这通常用于测试目的或计划维护期间。

消费者重组如何工作？

当发生重组时，组中的一个消费者（领导者）负责协调过程并确定如何将分区重新分配给组内所有活动的消费者。在各种情况下都会触发重组，例如消费者加入或离开组，或者主题中添加或删除了分区。

在本解释中，我们将详细探讨 Kafka 中的消费者重组如何工作，并提供一个使用自定义 ConsumerRebalanceListener 高效处理重组的 Kafka 消费者实现。

消费者重组的关键步骤

Kafka 中的重组过程通常涉及以下步骤

消费者加入或离开组：当消费者加入或离开组时，组协调器会触发重组。组协调器是负责管理组内消费者状态和分区分配的 Kafka broker。
领导者选举：组中的一个消费者被选为领导者。领导者的角色是协调重组过程，决定如何将分区分配给消费者，并将这些决定传达给组协调器和其他消费者。
分区分配：Kafka 提供了几种分区分配策略，领导者可以使用这些策略将分区分配给消费者。最常见的策略包括
- RangeAssignor：分区被分成连续的范围并分配给消费者。
- RoundRobinAssignor：分区在消费者之间平均分配。
- StickyAssignor：此策略尝试在重组期间保持相同分区分配，以最大限度地减少中断。
通信和同步：一旦确定了分区分配，领导者会将新的分配传达给组协调器。然后，组协调器将新分配通知给其他消费者。
分区分配通知：每个消费者都会收到其负责的分区的通知。此时，消费者可以开始从分配的分区处理消息。

触发消费者重组的因素

消费者重组由以下事件触发

消费者加入或离开组：当新消费者加入组时，组协调器会重新分配分区以包含新消费者。同样，如果消费者离开组（由于故障或手动关闭），其分区将被重新分配给剩余的消费者。
分区更改：如果主题中的分区数量发生变化（例如，如果向主题添加了新分区），组协调器将触发重组，将新分区重新分配给消费者。
心跳失败：每个消费者会定期向组协调器发送心跳，以表明它仍然活跃。如果组协调器在指定的时间间隔内未收到来自消费者的心跳，它会假设该消费者已失败并触发重组。

分区分配策略

如前所述，Kafka 使用不同的策略来分配分区给消费者。以下是最常见的策略

RangeAssignor：此策略将分区分成连续的块并分配给消费者。如果有 n 个分区和 m 个消费者，则第 i 个消费者被分配从 i * (n / m) 到 (i + 1) * (n / m) - 1 的分区。如果分区数量不能被消费者数量整除，此策略可能导致负载分配不均。
RoundRobinAssignor：在此策略中，分区在消费者之间平均分配。第一个分区分配给第一个消费者，第二个分区分配给第二个消费者，依此类推。一旦所有消费者都被分配了一个分区，下一个分区将再次分配给第一个消费者，循环重复。这种方法通常能确保更均衡的分区分配。
StickyAssignor：StickyAssignor 旨在最大限度地减少重组期间的分区重新分配。它尝试在重组期间保持分区分配给相同的消费者，这有助于减少重组导致的中断。
CooperativeStickyAssignor：这是 StickyAssignor 的高级版本，它确保了更具协作性的重组方法，允许在重组期间进行部分分区移动。这有助于最大限度地减少重组期间的停机时间。

ConsumerRebalanceListener 接口的作用

Kafka 允许开发者实现 ConsumerRebalanceListener 接口来更有效地处理重组事件。通过使用 ConsumerRebalanceListener 接口，开发者可以更精确地控制重组过程，并减少分区重新分配期间消息丢失或重复的可能性。

示例：具有自定义重组处理的 Kafka 消费者

现在让我们看一个使用 ConsumerRebalanceListener 高效处理重组事件的 Kafka 消费者完整示例

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

public class KafkaRebalanceConsumer {

    private static KafkaConsumer<String, String> consumer;
    private static Map<TopicPartition, Long> currentOffsets = new HashMap<>();

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        // Create a Kafka consumer instance
        consumer = new KafkaConsumer<>(properties);

        // Subscribe to the topic and attach the custom ConsumerRebalanceListener
        consumer.subscribe(Arrays.asList("my-topic"), new ConsumerRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                System.out.println("Partitions revoked: " + partitions);
                // Commit offsets for partitions being revoked
                consumer.commitSync(currentOffsets);
            }
            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                System.out.println("Partitions assigned: " + partitions);
                // Optionally reset the offsets to the latest committed values
                for (TopicPartition partition : partitions) {
                    consumer.seek(partition, currentOffsets.getOrDefault(partition, 0L));
                }
            }
        });

        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("Consumed record with key %s and value %s from partition %d and offset %d%n",
                            record.key(), record.value(), record.partition(), record.offset());

                    // Save the current offset for each partition
                    currentOffsets.put(new TopicPartition(record.topic(), record.partition()), record.offset() + 1);
                }

                // Periodically commit the offsets for the consumed records
                consumer.commitSync(currentOffsets);
            }
        } finally {
            consumer.close();
        }
    }
}

示例输出

自定义重组监听器：ConsumerRebalanceListener 用于在分区被撤销时提交偏移量，并在分区被重新分配时将消费者的位置设置到正确的偏移量。
撤销时同步提交：在 onPartitionsRevoked() 方法中，将提交被撤销分区的当前偏移量。这确保了在分区重新分配时不会丢失任何数据。
定位到最新偏移量：在 onPartitionsAssigned() 方法中，将消费者重新定位到每个新分配分区的正确偏移量。这确保了消费者从正确的点恢复消息消费。
偏移量管理：消费者维护一个每个分区当前偏移量（currentOffsets）的映射。当消费者处理消息时，此映射会被更新，并在重组期间用于提交偏移量和定位到正确的偏移量。

消费者重组监听器

Kafka 中的消费者重组监听器 (Consumer Rebalance Listener) 是一个强大的接口，它允许开发者在重组事件期间管理分区分配和撤销。Kafka 的默认重组机制会动态地将分区分配给消费者组中的消费者，并且当任何消费者加入或离开组时，分区都会被重新分配。虽然 Kafka 会自动处理此过程，但开发者可以通过实现 ConsumerRebalanceListener 接口来干预重组过程中的关键点。这使得消费者能够控制提交偏移量、为分区丢失做准备以及在重新分配后从正确的偏移量恢复消费。

这个主题尤为重要，因为重组会影响消息处理，如果管理不当，可能导致数据丢失、重复或延迟。通过使用 ConsumerRebalanceListener，开发者可以确保消费者能够优雅地处理分区撤销和分配，从而确保数据的一致性和正确性。

ConsumerRebalanceListener 接口的关键方法

ConsumerRebalanceListener 接口包含两个重要方法

onPartitionsRevoked(Collection<TopicPartition> partitions)：在消费者因重组而失去分区所有权之前调用此方法。在此方法中，消费者应提交它即将丢失的分区的偏移量，以避免数据丢失。在此处提交偏移量至关重要，因为 Kafka 会将这些分区重新分配给另一个消费者，而新消费者需要从正确的偏移量开始。
在 onPartitionsRevoked() 中要处理的步骤
- 提交偏移量：在释放分区之前，消费者应提交其拥有的所有分区的当前偏移量。这保证了新分配的消费者（重组后）将从正确的位置开始。
- 清理状态：如果应用程序为分区持有任何状态（例如，内存中的数据结构），它应该释放或清理该状态，因为它不再拥有这些分区的控制权。
onPartitionsAssigned(Collection<TopicPartition> partitions)：在重组完成后，并且消费者被重新分配了分区之后调用此方法。在此方法中，消费者应将其自身设置为从新分配分区的正确偏移量（通常是最后提交的偏移量）开始读取。
在 onPartitionsAssigned() 中要处理的步骤
- 定位到正确偏移量：一旦消费者被分配了新分区，它必须在开始消费之前定位到正确的偏移量。通常，这是每个分区的最后提交偏移量。Kafka 中的 seek() 方法用于将消费者对每个分子的位置设置为所需的偏移量。
- 恢复消费：设置了正确的偏移量后，消费者就可以从新分配的分区恢复消息处理。

示例程序：带有自定义重组监听器的 Kafka 消费者

下面是一个使用 ConsumerRebalanceListener 处理分区撤销和分配的 Kafka 消费者实现。

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
public class KafkaConsumerWithRebalanceListener {
    private static KafkaConsumer<String, String> consumer;
    private static Map<TopicPartition, Long> currentOffsets = new HashMap<>();

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group-id");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        consumer = new KafkaConsumer<>(properties);

        // Subscribe to the topic and attach a custom rebalance listener
        consumer.subscribe(Arrays.asList("my-topic"), new CustomRebalanceListener());

        // Poll messages from Kafka
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("Consumed record with key %s and value %s from partition %d at offset %d%n",
                            record.key(), record.value(), record.partition(), record.offset());
                    currentOffsets.put(new TopicPartition(record.topic(), record.partition()), record.offset() + 1);
                }

                // Periodically commit offsets for records that have been processed
                consumer.commitSync(currentOffsets);
            }
        } finally {
            consumer.close();
        }
    }

    // Custom rebalance listener implementation
    static class CustomRebalanceListener implements ConsumerRebalanceListener {

        @Override
        public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
            System.out.println("Partitions revoked: " + partitions);

            // Commit offsets for partitions that are about to be revoked
            consumer.commitSync(currentOffsets);
        }

        @Override
        public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
            System.out.println("Partitions assigned: " + partitions);

            // For newly assigned partitions, seek to the last committed offset
            for (TopicPartition partition : partitions) {
                Long offset = currentOffsets.get(partition);
                if (offset != null) {
                    consumer.seek(partition, offset);
                } else {
                    consumer.seekToBeginning(partitions);
                }
            }
        }
    }
}

示例输出

CustomRebalanceListener 类：此类实现了 ConsumerRebalanceListener 接口。它覆盖了两个方法（onPartitionsRevoked 和 onPartitionsAssigned）来处理重组事件。

onPartitionsRevoked：在此方法中，我们提交即将被撤销的分区的当前偏移量。这确保了分区偏移量在消费者失去其所有权之前被安全地提交。

onPartitionsAssigned：在重组完成后，并且分配了新分区后，此方法将定位到每个分区的最后提交偏移量（如果存在），或者定位到开头（如果未找到偏移量）。

currentOffsets Map：currentOffsets 映射存储每个分区的当前偏移量。它会在消息被消费时更新，并在重组期间使用，以确保消费者从正确的偏移量恢复处理。

提交和定位：commitSync 方法用于在分区被撤销之前提交偏移量，seek 方法用于在分区分配之后设置消费者的位置。这些方法确保了消费者在重组后从正确的位置继续处理。

在消费者重组中使用消费者重组监听器的重要性

使用 ConsumerRebalanceListener 提供了几个关键优势

偏移量管理
流处理中最关键的任务之一是管理偏移量。没有监听器，Kafka 会自动处理此问题，但在需要严格控制偏移量的情况下（例如，用于精确一次处理），重组监听器可确保消费者提交并从正确的偏移量恢复。
数据一致性
如果未在重组前正确提交偏移量，数据可能会被重新处理（导致重复）或完全丢失（导致数据丢失）。通过在 onPartitionsRevoked() 期间提交偏移量，消费者可以确保它不会不必要地丢失或重新处理数据。
状态管理
对于维护内存状态的应用程序，重组监听器是一个在分区被重新分配时清理或保留该状态的机会。这在有状态的流处理应用程序中尤为重要，在这些应用程序中，消费者需要随着时间的推移跟踪和聚合数据。
改进的容错能力
在像 Kafka 这样的分布式系统中，消费者故障很常见。通过正确管理分区分配和偏移量提交，重组监听器有助于增强系统的容错能力。

高级重组策略

虽然上面的示例演示了 ConsumerRebalanceListener 的直接实现，但 Kafka 也支持更高级的重组策略，例如

StickyAssignor
此策略旨在最大限度地减少重组期间消费者之间的分区移动。它试图使分区分配尽可能稳定，从而减少重组引起的开销。
协作重组
Kafka 的新重组协议允许部分重组，即在任何给定时间只重新分配一部分分区。这减少了与重组相关的停机时间，并提高了整体可用性。

下一主题Kafka-log-analysis

Kafka 消费者重新平衡

Kafka 中何时会发生重组

触发重组的关键场景

消费者重组如何工作？

消费者重组的关键步骤

触发消费者重组的因素

分区分配策略

ConsumerRebalanceListener 接口的作用

示例：具有自定义重组处理的 Kafka 消费者

消费者重组监听器

ConsumerRebalanceListener 接口的关键方法

示例程序：带有自定义重组监听器的 Kafka 消费者

在消费者重组中使用消费者重组监听器的重要性

高级重组策略

联系信息

关注我们

教程

面试题

在线编译器

Python

Java

.Net Framework

AI, ML and Data Science

Cloud Technology

B.Tech and MCA

Web Technology

PHP

Software Testing

Technical Interview

Java Interview

Python

Web Interview

Database Interview

B.Tech / MCA

Important Interview

Software Testing Interview

Company Interviews

Online Compilers

Multiple Choice Questions

Kafka 教程

Kafka 安装

Kafka CLI

Kafka 编程

实时示例

Kafka 监控

Kafka Connect

Kafka Streams

杂项

Kafka 消费者重新平衡

Kafka 中何时会发生重组

触发重组的关键场景

消费者重组如何工作？

消费者重组的关键步骤

触发消费者重组的因素

分区分配策略

ConsumerRebalanceListener 接口的作用

示例：具有自定义重组处理的 Kafka 消费者

消费者重组监听器

ConsumerRebalanceListener 接口的关键方法

示例程序：带有自定义重组监听器的 Kafka 消费者

在消费者重组中使用消费者重组监听器的重要性

高级重组策略

相关帖子

使用 Kafka 在 Cosmos DB 中处理大规模优化数据

事件流架构的用例

Kafka 中的事务性数据管道

Kafka 安全性

Kafka 复制

Kafka 在灾难恢复规划中的应用

用于物联网数据摄取管道的 Kafka

Kafka 连接器

Kafka 中的系统监控和警报

Kafka 作为智能城市的中枢神经系统

订阅 Tpoint Tech

联系信息

关注我们

教程

面试题

在线编译器