kafka java 小记录
本文章 方便日后检察使用。创建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开始消费
@Test
void 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报告已经消费，可以重复使用，
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
页:
[1]