kafka消息的多种实现（java泛型）[重复]

Question

我正在尝试根据事件类型为 kafka 事件创建多个实现类。

public class KafkaListener {
    @Autowired
    Service service;

    @KafkaListener(topics = ("mytopic"), containerFactory = "kafkaListenerContainerFactory")
    public void consumeSource(Object event) {
        service.process(event);
    }
}

public interface Service<E> {
    void process(E event);
}

public class ServiceImpl1 implements Service<Event1> {
    void process(Event1 event1) {
         // process 
    }
}

public class ServiceImpl2 implements Service<Event2> {
    void process(Event2 event2) {
         // process 
    }
}

//Event1 & Event2 are 2 POJO classes with different inputs

是否可以实现或者我应该创建多个侦听器，每个事件类型一个？

Answer 1

只要事件被Kafka反序列化，就可以使用类级别@KafkaListener和方法级别@KafkaHandlers。

见the documentation。

在类级别使用@KafkaListener 时，必须在方法级别指定@KafkaHandler。传递消息时，转换后的消息负载类型用于确定调用哪个方法。以下示例显示了如何执行此操作：

@KafkaListener(id = "multi", topics = "myTopic")
static class MultiListenerBean {

    @KafkaHandler
    public void listen(String foo) {
        ...
    }

    @KafkaHandler
    public void listen(Integer bar) {
        ...
    }

    @KafkaHandler(isDefault = true`)
    public void listenDefault(Object object) {
        ...
    }

}

从版本 2.1.3 开始，您可以指定一个 @KafkaHandler 方法作为默认方法，如果其他方法没有匹配则调用该方法。最多只能指定一种方法。使用@KafkaHandler methods 时，payload 必须已经转换为域对象（这样才能进行匹配）。使用自定义解串器 JsonDeserializer 或 JsonMessageConverter 并将其 TypePrecedence 设置为 TYPE_ID。有关详细信息，请参阅序列化、反序列化和消息转换。