我有多个生产者可以向一个Kafka主题发送多种类型的事件。
我还有一个消费者,必须消耗所有类型的消息,并为每种类型的消息使用不同的逻辑。
@KafkaListener(topics = "test", containerFactory = "kafkaListenerContainerFactory")
public void handleEvent(Message<EventOne> event) {
logger.info("event={}", event);
}
但是在这种情况下,所有的消息都会传递到这个方法中,而不仅仅是EventOne。
如果我为每种类型的消息实现两种方法,那么所有的消息都只会传递到一个方法中。
如果我像这样实现监听器:
@KafkaListener(topics = "test", containerFactory = "kafkaListenerContainerFactory")
public void handleEvent(Message<?> event) {
logger.info("event={}", event);
}
然后我遇到了异常: org.springframework.kafka.KafkaListenerEndpointContainer#0-0-kafka-listener-1] ERROR org.springframework.kafka.listener.LoggingErrorHandler - Error while processing: ConsumerRecord java.lang.IllegalArgumentException: Unrecognized Type: [null]
请告诉我如何实现多类型消费者?