本文章 方便日后查看使用。
创建kafka消费者（此处没啥可讲的，看注释吧。 手动版本的，更加灵活）
public KafkaConsumer<String, String> getCustomer() {// 1. 配置属性参数Properties properties = new Properties();// 设置Kafka集群的地址和端口，消费者将连接到这个地址和端口properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");// 设置键（Key）的反序列化器为StringDeserializer，用于将字节数据转换为String类型properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());// 设置值（Value）的反序列化器为StringDeserializer，用于将字节数据转换为String类型properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());// 设置消费者所属的消费者组，消费者组内的消费者将共同消费同一个Topic的消息properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");// 设置消费者与Kafka集群之间的会话超时时间（单位：毫秒）properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10000);// 设置消费者是否自动提交offset，true表示自动提交properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);// 设置自动提交offset的时间间隔（单位：毫秒）properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 5000);// 设置每次poll操作返回的最大记录数properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,5);// 根据配置属性创建Kafka消费者实例return new KafkaConsumer<>(properties);}使用kafka开始消费
@Testvoid KafkaConsumerTest() {KafkaConsumer<String, String> consumer = kafkaNdpiCustomer.getCustomer();// 订阅要消费的主题，这里是 "test Topic"consumer.subscribe(Collections.singletonList("test-Topic"));ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));for (ConsumerRecord<String, String> record : records) {// 处理消息的逻辑System.out.printf("消息为：offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
// boolean flag = true;
// if (flag){
// consumer.commitAsync();
// break;
// }}consumer.close();System.out.println("结束消费");}其中 consumer.commitAsync(); 为异步提交，不选择则不会像kafka报告已经消费，可以重复使用，
