java - spring Kafka 负载类型的模糊方法

Question

我尝试使用 spring-kafka-2.2.0 和 spring-boot-2.1.0 使用 Kafka 主题中的两种不同类型的有效负载，但遗憾的是没有到目前为止，运气都不错。

所以我的项目中有两个模型 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

Answer 1

@KafkaHandler 方法不适用于 BATCH 模式。至少以您现在所拥有的方式:没有方法可以通过通用参数来处理不同的批处理。

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

注意:Apache Kafka 不会在不同批处理之间分发反序列化记录。它们都将在同一个 ConsumerRecords 中为监听器生成。因此，尝试按列表中的泛型类型进行分发的逻辑永远不会发生，即使我们会找出方法参数的泛型类型......因此，简单的 RECORD 更多是你的救星。