使用consumer.poll写一段在java中kafka消费代码，要避免数据漏消费

在Java中使用Kafka消费者`Consumer.poll()`来处理消息，为了避免数据漏消费，你需要设置合适的消费组和超时时间，并且确保消费者的运行状态良好。下面是一个基本示例：

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

public class KafkaConsumerExample {
    private final String bootstrapServers = "localhost:9092"; // 替换为你服务器的实际地址
    private final String groupId = "my-consumer-group"; // 定义消费组名
    private final String topicName = "my-topic"; // 指定你要消费的主题

    public void consumeMessages() {
        Properties props = new Properties();
        props.put("bootstrap.servers", bootstrapServers);
        props.put("group.id", groupId);
        props.put("enable.auto.commit", "true"); // 自动提交偏移量
        props.put("auto.commit.interval.ms", "5000"); // 提交间隔，单位毫秒

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(100); // 设置最大等待时间（这里是100ms）
                
                for (ConsumerRecord<String, String> record : records) {
                    // 处理接收到的消息
                    System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
                    
                    // 可能需要确认消费（这里仅作演示，实际生产环境中可能会有业务处理逻辑）
                    consumer.commitSync();
                }
            }
        } finally {
            consumer.close();
        }
    }

    public static void main(String[] args) {
        KafkaConsumerExample consumer = new KafkaConsumerExample();
        consumer.consumeMessages();
    }
}

在这个例子中，我们设置了自动提交偏移量并定期检查新消息。如果程序中断，Kafka会自动从上一次消费的位置继续。但是，如果你担心网络延迟或其他异常导致消息丢失，可以考虑增加重试机制或者监控系统的可靠性。