使用来自同一kafka主题的字符串和JSON消息-反序列化问题

Question

JSON消息正在被字符串使用者消费。My Products发送两种类型的消息字符串和序列化JSON

@KafkaListener(topics = "${kafka.topic}", groupId = "${kafka.group-id}", containerFactory = "${kafka.string-listener-container-factory}")
public void consume(@NotNull ConsumerRecord<String, String> cr, @Payload String payload) {
    log.debug("Received asset id: {}, with key: {}, Partition: {}, Offset: {}  ", payload, cr.key(), cr.partition(),
            cr.offset());

}


@KafkaListener(topics = "${kafka.topic}", groupId = "${kafka.group-id}", containerFactory = "${kafka.json-listener-container-factory}")
public void consumeAssetEvent(@NotNull ConsumerRecord<String, Event> cr, @Payload Event payload) {
    log.debug("Received asset id: {}, with key: {}, Partition: {}, Offset: {}  ", payload, cr.key(), cr.partition(),
            cr.offset());

}

消费者方面我有两个消费者1。正在侦听字符串消息%2。监听json并反序列化为对象

即使是json消息也由String Listener使用。

Answer 1

在消费者中没有过滤功能，可以根据您设置的Serde（例如string、json等）来区分消息。当生产者发送一个消息时，它被转换为kafka主题中的Byte[]。然后由使用者反序列化设置反序列化该字节[]。

因此没有默认的方法来过滤字符串消息到字符串使用者和json到json使用者。或者创建listener，接收所有数据并检查它是否是json，或者更改string和json的主题（将string发送到一个主题，将json发送到另一个主题并相应地使用）。