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报告已经消费,可以重复使用,
原文地址:https://blog.csdn.net/Mi_RenYu/article/details/144373290
免责声明:本站文章内容转载自网络资源,如本站内容侵犯了原著者的合法权益,可联系本站删除。更多内容请关注自学内容网(zxcms.com)!