时间戳

首页 > 解决方案 > 有效载荷类型的弹簧卡夫卡模糊方法

java - 有效载荷类型的弹簧卡夫卡模糊方法

问题描述

我正在尝试使用来自 Kafka 主题的两种不同类型的有效负载spring-kafka-2.2.0，spring-boot-2.1.0但不幸的是，到目前为止没有运气异常。

所以我的项目中有两个模型，com.kafka.model.Professor并且com.kafka.model.Student我也在类型映射中配置了它们。但我不确定我在这个反序列化过程中遗漏了哪一部分。

配置

@Configuration
public class KafkaConsumerConfig {

@Value("${kafka.consumer.bootstrap-server}")
private String bootstrapServer;

@Bean("consumerConfigs")
public Map<String, Object> consumerConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-4");
    props.put(JsonDeserializer.TYPE_MAPPINGS,
            "professor:com.kafka.model.Professor, student:com.kafka.model.Student");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
    props.put(ConsumerConfig.CLIENT_ID_CONFIG, "test-4");
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "1");
    return props;
}

@Bean("consumerFactory")
public ConsumerFactory<String, Object> consumerFactory() {
    JsonDeserializer<Object> jsonDeserializer = new JsonDeserializer<>(Object.class);
    jsonDeserializer.addTrustedPackages("*");

    return new DefaultKafkaConsumerFactory<>(consumerConfigs(), new StringDeserializer(), jsonDeserializer);
}

@Bean("kafkaListenerContainerFactory")
@ConditionalOnMissingBean(name = "KafkaAutoConfiguration")
public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setBatchListener(true);
    factory.getContainerProperties().setAckMode(AckMode.BATCH);
    return factory;
    }

}

监听类

@Service
@KafkaListener(topics = "test-events",id = "kafkaListenerContainerFactory-11")
public class KafkaConsumerService {

@KafkaHandler
public void student(List<Student> stu) {
    System.out.println(stu);
}

@KafkaHandler
public void professor(List<Professor> pro) {
    System.out.println(pro);
   }


}

错误

2019-02-15 13:04:33.060 ERROR 14863 --- [actory-11-0-C-1] o.s.k.listener.BatchLoggingErrorHandler  : Error while processing:
ConsumerRecord(topic = stores-pricing-easytest-manual-events, partition = 4, offset = 1189400, CreateTime = 1550096019845, serialized key size = 7, serialized value size = 49, headers = RecordHeaders(headers = [], isReadOnly = false), key = student, value = {studentName=thomas, rollNum=109, age=35})

 org.springframework.kafka.KafkaException: Ambiguous methods for payload type: class java.util.ArrayList: student and professor

标签： javaspring-bootapache-kafkaspring-kafka

解决方案

该@KafkaHandler方法不适用于BATCH模式。至少在您现在拥有的方式上：没有任何方法可以通过通用参数来处理区分批次。

考虑关闭批处理模式或仅在单个@KafkaListener方法中处理所有内容，该方法已经接受 aList作为带有任何通用信息的有效负载的参数。

注意：Apache Kafka 不会在不同批次之间分发反序列化记录。它们都将在同一个中为听众制作ConsumerRecords。因此，您尝试通过列表中的泛型类型进行分配的逻辑永远不会发生，即使我们会找出方法参数的泛型类型......所以，更简单RECORD的是您的保护程序。

推荐阅读