kafka java 小记录

本文章 方便日后检察使用。

创建kafka消费者（此处没啥可讲的，看表明吧。 手动版本的，更加灵活）

1.     public KafkaConsumer<String, String> getCustomer() {
2.         // 1. 配置属性参数
3.         Properties properties = new Properties();
5.         // 设置Kafka集群的地址和端口，消费者将连接到这个地址和端口
6.         properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
8.         // 设置键（Key）的反序列化器为StringDeserializer，用于将字节数据转换为String类型
9.         properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
10.         // 设置值（Value）的反序列化器为StringDeserializer，用于将字节数据转换为String类型
11.         properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
13.         // 设置消费者所属的消费者组，消费者组内的消费者将共同消费同一个Topic的消息
14.         properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");
15.         // 设置消费者与Kafka集群之间的会话超时时间（单位：毫秒）
16.         properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10000);
17.         // 设置消费者是否自动提交offset，true表示自动提交
18.         properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
19.         // 设置自动提交offset的时间间隔（单位：毫秒）
20.         properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 5000);
21.         // 设置每次poll操作返回的最大记录数
22.         properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,5);
24.         // 根据配置属性创建Kafka消费者实例
25.         return new KafkaConsumer<>(properties);
26.     }

复制代码

使用kafka开始消费

1.     @Test
2.     void KafkaConsumerTest() {
3.         KafkaConsumer<String, String> consumer = kafkaNdpiCustomer.getCustomer();
5.         // 订阅要消费的主题，这里是 "test Topic"
6.         consumer.subscribe(Collections.singletonList("test-Topic"));
9.         ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
10.         for (ConsumerRecord<String, String> record : records) {
11.                 // 处理消息的逻辑
12.                 System.out.printf("消息为：offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
13. //                boolean flag = true;
14. //                if (flag){
15. //                    consumer.commitAsync();
16. //                    break;
17. //                }
18.         }
19.         consumer.close();
20.         System.out.println("结束消费");
21.     }

复制代码

其中  consumer.commitAsync(); 为异步提交，不选择则不会像kafka报告已经消费，可以重复使用，

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。