你刚才java kafka 消费者的代码没写全

非常抱歉，可能是我在回答时出现了错误。下面是一个完整的Java Kafka消费者示例代码： ```java import java.util.Collections; import java.util.Properties; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.common.serialization.StringDeserializer; public class KafkaConsumerExample { public static void main(String[] args) { // 设置Kafka消费者配置 Properties props = new Properties(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group"); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); // 创建Kafka消费者 KafkaConsumer consumer = new KafkaConsumer<>(props); // 订阅主题 consumer.subscribe(Collections.singletonList("test-topic")); // 消费消息 while (true) { ConsumerRecords records = consumer.poll(100); for (ConsumerRecord record : records) { System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value()); } } } } ``` 在这个示例中，我们首先设置了Kafka消费者的配置，包括Kafka服务器地址、消费者组ID、键和值的反序列化器等。 然后，我们创建了一个Kafka消费者对象，并订阅了一个名为“test-topic”的主题。 最后，我们使用一个无限循环来消费消息。在每次循环中，我们调用`poll()`方法从Kafka服务器获取一批消息，并遍历这些消息并打印它们的偏移量、键和值。 需要注意的是，Kafka消费者是线程不安全的，因此应该在单个线程中使用。如果您需要在多个线程中