如何使用 spring-kafka 框架将对象列表发送到 kafka 主题？答案

【问题标题】：How to send List of Objects to kafka topic using spring-kafka framework?如何使用 spring-kafka 框架将对象列表发送到 kafka 主题？
【发布时间】：2019-11-03 08:20:35
【问题描述】：

我在 spring-boot 应用程序中使用 spring-kafka 来发送数据主题。我需要从 oracle 表中获取数据并发送。

我从 oracle 表中获取列表。如何将它们发送到主题？

即

有没有办法将它们作为列表发送？如果是的话怎么办？如果是，那么如何在消费者端反序列化它？
是否可以使用 spring-book 和 spring-kafka 像流媒体方式一样发送数据？如果是，请提供更多信息或示例/sn-p ...

如果我一次发送List，partitionKey如何处理？

目前我正在发送单个公司 obj，因此密钥定义如下

companyKafkaTemplate.send(COMPANY_TOPIC,this.getKey(company), company);

【问题讨论】：

为什么不能使用 Kafka Connect 将数据库行发送到 Kafka？如果您绝对必须编写自己的生产者代码，请不要将行列表作为一条消息发送给 Kafka
@cricket_007，谢谢，我有另一个包装器/项目，它从底层数据库表中获取数据，一旦我拉出我需要使用 spring-kafka 发送方法发送的记录/对象列表，所以寻求建议如何做..如何与主题平行的多个记录，需要任何演员框架或弹簧就足够了？

标签： java spring-boot apache-kafka kafka-producer-api spring-kafka

【解决方案1】：

对于List 序列化和反序列化，我建议在 Spring Kafka 中使用JSON 支持：https://docs.spring.io/spring-kafka/docs/2.2.7.RELEASE/reference/html/#serdes

对于流，我建议查看 Spring Kafka 中的响应式支持，基于 Reactor Kafka 项目：https://github.com/reactor/reactor-kafka

为此，我们提供了ReactiveKafkaProducerTemplate 和ReactiveKafkaConsumerTemplate。

【讨论】：

谢谢，你的意思是对象列表？还是应该包裹到另一个对象中？如果是对象列表，那么如何对它们进行 serde 处理？
这足以作为列表发送，无需任何包装。不确定我关于 JSON（反）序列化的建议有什么问题......
给定链接docs.spring.io/spring-kafka/docs/2.2.7.RELEASE/reference/html/… 我没有看到任何支持列表的发送方法版本，即将列表作为参数。所以我有点困惑。在普通的 kafka 中我们有，但在 spring-kafka 中似乎没有
我很困惑。您只需将KafkaTemplate.send() 与List<?> 一起使用。 JsonSerializer 负责正确的列表及其项目序列化。 “我们拥有普通的 Kafka”是什么意思？
Kafka 没有区分单条记录和列表。该列表也将被视为单个记录。因此，您需要迭代您的列表并为该列表中的每个项目使用KafkaTemplate.send()。 Kafka 客户端能够缓冲生产者中的记录，然后一次性刷新它们以获得更好的性能，但它仍然不是关于发送单个列表的最终用户 API。因此，您的问题与 Apache Kafka API 无关：这是不可能的。我认为这不是故意的，因为 partitionKey 功能。