Kafka分组后的重新分配

以下是一个示例代码，展示了如何使用Kafka的消费者分组和重新分配功能。

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.util.Arrays;
import java.util.Map;
import java.util.Properties;

public class KafkaRebalanceExample {

    public static void main(String[] args) {
        // 配置Kafka消费者属性
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test-group");

        // 创建Kafka消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        // 订阅主题
        consumer.subscribe(Arrays.asList("test-topic"), new ConsumerRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                // 在重新分配之前的回调方法，可以用于提交偏移量
                // 在这个例子中，我们简单地打印出被撤销的分区
                System.out.println("Partitions revoked: " + partitions);
            }

            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                // 在重新分配之后的回调方法，可以用于恢复偏移量
                // 在这个例子中，我们简单地打印出被分配的分区
                System.out.println("Partitions assigned: " + partitions);
            }
        });

        try {
            while (true) {
                // 拉取消息
                ConsumerRecords<String, String> records = consumer.poll(100);

                // 处理消息
                for (ConsumerRecord<String, String> record : records) {
                    System.out.println("Received message: " + record.value());
                }

                // 手动提交偏移量
                consumer.commitSync();
            }
        } finally {
            // 关闭消费者
            consumer.close();
        }
    }
}

在这个示例中，我们创建了一个Kafka消费者，并订阅了一个名为"test-topic"的主题。我们还实现了ConsumerRebalanceListener接口，在重新分配之前和之后提供了回调方法。在回调方法中，我们可以执行一些操作，例如提交偏移量或恢复偏移量。

请注意，这只是一个示例代码，实际使用时可能需要根据具体需求进行修改和调整。