Kafka consumer频繁reblance

要创建一个Kafka Consumer，需要执行以下步骤： 1. 导入所需的 Kafka 相关库，如 kafka-clients 和 slf4j。 2. 配置 Consumer 的属性，例如 Kafka 服务器的地址、Consumer 组 ID、是否自动提交偏移量等。 3. 创建一个 KafkaConsumer 实例，将上一步中的属性传递给它。 4. 订阅一个或多个主题，使用 KafkaConsumer 的 subscribe() 方法。 5. 循环从 Consumer 中拉取消息，使用 KafkaConsumer 的 poll() 方法。 6. 处理从 Kafka 中拉取的消息。 下面是一个基本的 Kafka Consumer 的示例代码： ```java import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.common.serialization.StringDeserializer; import java.util.Collections; import java.util.Properties; public class MyKafkaConsumer { public static void main(String[] args) { // 配置 Consumer 的属性 Properties props = new Properties(); props.setProperty("bootstrap.servers", "localhost:9092"); props.setProperty("group.id", "my-group"); props.setProperty("enable.auto.commit", "true"); props.setProperty("auto.commit.interval.ms", "1000"); props.setProperty("key.deserializer", StringDeserializer.class.getName()); props.setProperty("value.deserializer", StringDeserializer.class.getName()); // 创建 KafkaConsumer 实例 KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); // 订阅一个主题 consumer.subscribe(Collections.singletonList("my-topic")); // 循环从 Consumer 中拉取消息 while (true) { ConsumerRecords<String, String> records = consumer.poll(100); // 处理从 Kafka 中拉取的消息 for (ConsumerRecord<String, String> record : records) { System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value()); } } } } ``` 注意，这只是一个基本的示例，实际使用中还需要考虑更多的细节和配置。