snailclimb / springboot-kafka Goto Github PK

您好，我最近一个参加比赛的项目中用到了springboot集成kafka。但在循环消费的部分，我的idea对如下语句报出了错误：for (ConsumerRecord<String, String> record : records) {，具体报错为：foreach 不适用于类型 'com.entor.service.ConsumerRecords<java.lang.String,java.lang.String>'。我在网上看到大家基本上都是这么写的，但不太明白为什么在我这里会报错。我曾经认为是Java版本的问题，但我从jdk从1.8换到15后也并没有解决问题。想请教一下这块儿为什么会报错。下面会附上部分源码，或许能方便您找到问题所在。
@Override public void run(){ //加载kafka消费者参数 Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("group.id", "ytna"); props.put("enable.auto.commit", "true"); props.put("auto.commit.interval.ms", "1000"); props.put("session.timeout.ms", "15000"); props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); //创建消费者对象 consumer = new KafkaConsumer<String,String>(props); consumer.subscribe(Arrays.asList(this.topic)); //死循环，持续消费kafka while (true){ try { //消费数据，并设置超时时间 ConsumerRecords<String, String> records = consumer.poll(100); //Consumer message for (ConsumerRecord<String, String> record : records) { //Send message to every client Iterable<? extends webSocket.WebSocket> wbSockets = null; for (com.entor.service.webSocket.WebSocket webSocket :wbSockets){ webSocket.sendMessage(record.value()); } } }catch (IOException e){ System.out.println(e.getMessage()); continue; } } }
感谢！！！

行：在 application.xml 配置文件中配置 Kafka 连接信息以及我们项目中用到的 topic

application.xml 应为 application.yaml

面试官问我如何保证Kafka不丢失消息?我哭了！ 点击不存在
期待 kafka的文章！