如何使用 Header 从一条 Kafka 消息中获取多个对象

Question

我想反序列化一条消息（来自同一主题）中的不同对象，并根据对象类型将其保存/更新到数据库中的适当表中。正如 Gary 提到的here 我可以使用标头来提供我的对象必须反序列化为哪种类型的信息。您能否提供一些示例来说明如何实现这一目标？

Answer 1

参见 sample-02 here。

示例 2

这个示例演示了一个简单的生产者和一个多方法的消费者；生产者发送Foo1 和Bar1 类型的对象，消费者接收Foo2 和Bar2 类型的对象（对象具有相同的字段foo）。

生产者使用JsonSerializer；消费者使用ByteArrayDeserializer 和ByteArrayJsonMessageConverter ，后者转换为侦听器方法参数的所需类型。 在这种情况下我们无法推断类型（因为类型用于选择要调用的方法）。 因此，我们在生产者和消费者端配置类型映射。 请参阅生产者端的 application.yml 和消费者端的 converter bean。

MultiMethods@KafkaListener有3个方法；一个用于每个已知对象，一个用于其他对象的后备默认方法。

运行应用程序并使用 curl 发送一些数据：

$ curl -X POST http://localhost:8080/send/foo/bar

$ curl -X POST http://localhost:8080/send/bar/baz

$ curl -X POST http://localhost:8080/send/unknown/xxx

控制台：

Received: Foo2 [foo=bar]

Received: Bar2 [bar=baz]

Received unknown: xxx