import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.TopicPartition;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Properties;

public class KafkaConsumerWithRebalanceListener {

    private static final String BOOTSTRAP_SERVERS = "localhost:9092"; 
    private static final String GROUP_ID = "my-consumer-group";
    private static final String TOPIC_NAME = "my-kafka-topic";

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        
        // "配置分配策略（默认是 StickyAssignor）"
        // props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, 
        //     "org.apache.kafka.clients.consumer.RangeAssignor");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        try {
            // "订阅 Topic"
            consumer.subscribe(Collections.singletonList(TOPIC_NAME), 
                new ConsumerRebalanceListener() {
                    @Override
                    public void onPartitionsRevoked(Collection&lt;TopicPartition&gt; partitions) {
                        System.out.println("【Rebalance 事件】分区被撤销");
                        for (TopicPartition partition : partitions) {
                            System.out.println("  撤销分区: " + partition);
                        }
                        // "在失去分区前，提交已处理消息的偏移量"
                        consumer.commitSync();
                    }

                    @Override
                    public void onPartitionsAssigned(Collection&lt;TopicPartition&gt; partitions) {
                        System.out.println("【Rebalance 事件】分区被分配");
                        for (TopicPartition partition : partitions) {
                            System.out.println("  新分配分区: " + partition);
                        }
                    }
                });

            System.out.println("消费者已启动，加入消费者组: " + GROUP_ID);

            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("【消息】分区: %d, 偏移量: %d, Key: %s, Value: %s%n",
                            record.partition(), record.offset(), record.key(), record.value());
                    
                    // "处理消息业务逻辑..."
                }
                
                // "批量处理完成后提交偏移量"
                if (!records.isEmpty()) {
                    consumer.commitSync();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            consumer.close();
            System.out.println("消费者已关闭");
        }
    }
}

7.2 "多消费者实例模拟"

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class MultiConsumerSimulation {

    private static final String BOOTSTRAP_SERVERS = "localhost:9092";
    private static final String GROUP_ID = "multi-consumer-group";
    private static final String TOPIC_NAME = "my-kafka-topic";
    private static final int NUM_CONSUMERS = 3; // "启动 3 个消费者"

    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(NUM_CONSUMERS);

        // "启动多个消费者线程"
        for (int i = 0; i < NUM_CONSUMERS; i++) {
            final int consumerId = i;
            executor.submit(() -> runConsumer(consumerId));
        }

        // "保持应用运行"
        try {
            Thread.sleep(Long.MAX_VALUE);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private static void runConsumer(int consumerId) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        try {
            consumer.subscribe(Collections.singletonList(TOPIC_NAME));
            System.out.println("消费者 " + consumerId + " 已启动");

            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("【消费者 %d】分区: %d, 偏移量: %d, 消息: %s%n",
                            consumerId, record.partition(), record.offset(), record.value());
                    
                    // "模拟处理延迟"
                    Thread.sleep(100);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            consumer.close();
            System.out.println("消费者 " + consumerId + " 已关闭");
        }
    }
}

7.3 "运行与观察"

"Step 1: 创建 Topic"

# "创建一个 6 个分区的 Topic"
kafka-topics.sh --bootstrap-server localhost:9092 --create --topic my-kafka-topic --partitions 6 --replication-factor 1

"Step 2: 启动消费者"

# "运行 MultiConsumerSimulation，会启动 3 个消费者"
# "观察控制台输出，会看到 Rebalance 过程"

"Step 3: 启动生产者发送消息"

# "使用生产者发送消息到 Topic"
kafka-console-producer.sh --bootstrap-server localhost:9092 --topic my-kafka-topic
# "输入消息后按 Enter 发送"

"预期观察结果"：

"消费者启动时会经历 Rebalance，分配 6 个分区给 3 个消费者。"
"每个消费者分配 2 个分区，如 C0 分配 P0、P3，C1 分配 P1、P4，C2 分配 P2、P5。"
"发送消息时，根据消息 Key 的哈希值分配到不同的分区，由对应的消费者处理。"

8) "分配过程中的关键配置"

8.1 "消费者关键配置"

[嵌入内容: diagram]

8.2 "配置最佳实践"

Properties props = new Properties();

// "关键配置"
props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");

// "分配策略（选择其中一个）"
props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, 
    "org.apache.kafka.clients.consumer.StickyAssignor"); // "推荐"

// "心跳和会话配置"
props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 3000); // "3 秒发送一次心跳"
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10000); // "10 秒超时"
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000); // "5 分钟内必须 poll 一次"

// "偏移量和提交配置"
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); // "建议手动提交"

// "性能配置"
props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1024); // "最小 1KB"
props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 500); // "最多等待 500ms"
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500); // "一次 poll 最多 500 条消息"