带有Avro和Schema Repo的apachekafka-Schema Id在消息中的位置？

Question

我想使用Avro来序列化我的Kafka消息的数据，并想将其与Avro模式存储库一起使用，这样我就不必将模式包含在每条消息中。

将Avro与Kafka结合使用似乎是一件很流行的事情，许多博客/堆栈溢出问题/用户组等都提到了将模式Id与消息一起发送，但我找不到一个实际的示例来说明它应该去哪里。

我想它应该放在Kafka消息头的某个地方，但我找不到一个明显的地方。如果它在Avro消息中，则必须根据模式对其进行解码，以获取消息内容并显示需要解码的模式，这显然存在问题。

我正在使用C#客户端，但任何语言的示例都很好。message类具有以下字段：

public MessageMetadata Meta { get; set; }
public byte MagicNumber { get; set; }
public byte Attribute { get; set; }
public byte[] Key { get; set; }
public byte[] Value { get; set; }

但这些似乎都不正确。MessageMetaData只有Offset和PartitionId。

那么，Avro模式Id应该放在哪里呢？

Answer 1

模式id实际上是在avro消息本身中编码的。看看这个，看看编码器/解码器是如何实现的。

一般来说，当您向Kafka发送Avro消息时会发生什么：

1. 编码器从要编码的对象中获取模式。
2. 编码器向架构注册表询问此架构的id。如果架构已经注册，您将获得一个现有id，如果没有-注册表将注册架构并返回新id。
3. 对象的编码如下：[魔法字节][模式ID][实际消息]其中魔法字节只是一个用于区分此类消息的0x0字节，模式ID是一个4字节的整数值其余的是实际编码的消息。

当您解码信息时，会发生以下情况：

1. 解码器读取第一个字节并确保其为0x0
2. 解码器读取接下来的4个字节并将其转换为整数值。这就是模式id的解码方式
3. 现在，当解码器有一个模式id时，它可能会向模式注册表询问该id的实际模式。瞧

如果您的密钥是Avro编码的，那么您的密钥将采用上述格式。值也是如此。这样，您的键和值可能都是Avro值并使用不同的模式。

编辑以回答评论中的问题：

实际的模式存储在模式存储库中（这实际上是模式存储库的整个点，用于存储模式：）。Avro对象容器文件格式与上述格式无关。Kafka夫罗编码器/解码器使用略有不同的消息格式（但实际消息的编码方式完全相同）。

这些格式之间的主要区别在于，对象容器文件携带实际的模式，并且可能包含与该模式对应的多条消息，而上面描述的格式只携带模式id，并且只携带与该模式对应的一条消息。

传递对象容器文件编码的消息可能不容易理解/维护，因为一条Kafka消息将包含多条Avro消息。或者您可以确保一条Kafka消息只包含一条Avro消息，但这会导致每个消息都带有模式。

Avro模式可能非常大（我见过600 KB或更大的模式），将模式与每条消息一起携带会非常昂贵和浪费，因此这就是模式存储库的作用所在-模式只提取一次，并在本地缓存，所有其他查找都只是快速的映射查找。