kafka消费消息的时候，报错No converter found capable of converting from type [java.lang.String] to type [org. apache.kafka.clients.consumer.ConsumerRecord<? ?>，
没有消费到数据，这种情况可能是发送方发送的数据是封装了多个ConsumerRecord<? ?>对象发送过来的，需要用ConsumerRecords<? ?> 来接收，该对象与前面对象相比后面多了一个 s ，通过对ConsumerRecords<? ?>遍历可以获取多个ConsumerRecord<?, ?>对象，具体代码如下：
import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
@Component
public class Test{
//多个ConsumerRecord<?, ?>对象接收
@KafkaListener(topics = {"TestTopic01"})
public void consumerMessage01(ConsumerRecords<?, ?> message) {
Iterator<? extends ConsumerRecord<?, ?>> iterator = message.iterator();
while (iterator.hasNext()){
ConsumerRecord<?, ?> next = iterator.next();
Optional optional =Optional.ofNullable(next.value());
//使用 isPresent() 方法判断该对象是否包含非 null 的值
if(optional.isPresent()){
Object msg =optional.get();
//TestEntity是业务中使用到的实体对象
TestEntity testEntity = JSON.parseObject(msg.toString(),TestEntity.class);
}
}
}
//单个ConsumerRecord<?, ?>对象接收
@KafkaListener(topics = {"testTopic02"})
public void consumerMessage02(ConsumerRecord<?, ?> message) {
Optional optional =Optional.ofNullable(message.value());
Object msg =optional.get();
TestEntity testEntity = JSON.parseObject(msg.toString(),TestEntity.class);
}
————————————————
版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。
原文链接：https://blog.csdn.net/k0307x1990y/article/details/130356983