带有领域事件的 Kafka

Question

在我的事件驱动项目中，我有 Commands 类型的消息，作为响应，我有 Events。

这些Commands 和Events 消息表示域，因此它们包含域中的复杂类型。

例子：

RegisterClientCommand(Name, Email)

ClientRegisteredEvent(ClientId)

域中还有数十个这样的命令和事件对。

我在想这样的事情：

RawMessage(payloadMap, sequenceId, createdOn)

有效负载将包含消息域类类型名称和消息字段。

我也在阅读 Avro 格式，但似乎需要为每条消息定义消息格式做很多工作。

就实际通过 Kafka 代理传输的消息格式而言，最佳实践是什么？

Answer 1

没有单一的“最佳”方式来做到这一点，这完全取决于您的团队/组织的专业知识以及您项目的具体要求。

Kafka 本身对消息实际包含的内容漠不关心。大多数时候，它只是将消息值和键视为不透明的字节数组。

无论您最终将RawMessage 定义为Java 端的什么，都必须将其序列化为字节数组才能将其生成到Kafka，因为这是KafkaProducer 所要求的。也许它是您已经拥有的自定义字符串序列化器，也许您可​​以使用 Jackson 或类似的东西将 POJO 序列化为 JSON。或者，也许您只是发送一个巨大的逗号分隔字符串作为消息。这完全取决于您。

重要的是，消费者在从 kafka 主题拉取消息时，能够正确可靠地从消息中的每个字段读取数据，没有任何错误、版本冲突等。大多数 serde/schema 机制存在的，如 Avro、Protobuf 或 Thrift，尝试让这项工作对你来说更容易。尤其是复杂的事情，例如确保新消息与同一消息的先前版本向后兼容。

• 大多数人最终会遇到以下几种情况：
  • 用于创建字节数组以生成到 Kafka 中的 Serde 机制，一些流行的机制是 Avro、Protobuf、Thrift。
  • 原始 JSON 字符串
  • 具有某种内部/自定义格式的巨大字符串，需要对其进行解析/分析。
• 一些公司使用集中式架构服务。这样您的数据消费者不必提前知道消息包含什么模式，他们只需拉下消息，然后从服务中请求相应的模式。 Confluent 有自己的自定义模式注册表解决方案，多年来一直支持 Avro，并且在几周前，现在正式支持 Protobuf。这不是必需的，如果您拥有端到端的生产者/消费者，您可能会决定自己处理序列化，但很多人已经习惯了。
• 根据消息类型，有时您需要压缩，因为消息可能非常重复和/或很大，因此如果您发送压缩消息，最终会节省相当多的存储空间和带宽，但会消耗一些 CPU使用和延迟。这也可以在生产者/消费者端由您自己处理，在字节数组被序列化后压缩它们，或者您可以直接在生产者端请求消息压缩（在 Kafka 文档中查找 compression.type）。李>