kafka 协议分析 (一) 基础篇
kafka 协议分析 (二) Produce API
kafka 协议分析 (三) Fetch API
Kafka作为分布式消息组件被广泛使用，官方也提供了非常易用的各语言版本的client。由于系统的异构性，我们使用了不同语言的client生产和消费消息。实际使用的过程中，发现有生产了消息却消费不到等情况。由于client屏蔽了很多底层细节，所以通过网络抓包去分析数据才是解决问题的关键。要分析网络数据包就必须了解KAFKA PROTOCOL。（PS：本文是假定你对Kafka的基本概念已经了解，特别是topic,partition,broker和leader）
一个网络包样例
image.png
协议基础
协议：为了实现一个目的，怎样构建请求以及如何解析响应。
要想分析和使用kafka协议，我们必须要事先弄清楚协议的基础理论。本文主要介绍这些基础概念，后面文章会陆续介绍kafka的常用api(主要介绍low level api和high level api)。
TCP
- kafka协议是基于tcp的
- 消息体全部由后面介绍的3种类型的数据组成
- 对于同一tcp链接，服务端会按序响应请求（所以client应采用NIO模型以提高效率）
Partition
- topic里的数据是被分配到多个partition上存储的，每个partition又会有N-1个备份(注意:N为备份因子,N>=1,N＝1表示没有备份)。将数据分片和并备份存储，即提高了效率和又保证了安全性，类似理论在大部分的分布式系统中都有体现。
- 数据要被配到哪个partition上存储，是由生产者客户端(producer client)决定的，而不是kafka服务器(broker)决定的。这种设计使得开发者自由度更高，另一方面也要求客户端必须决定分区策略，大部分情况下会使用：（1）生产者可以通过轮询或随机的方式向所有partition写入数据，以保证数据被平均分配到了每个partition上。（2）生产者也可以通过数据的Key值来计算(例如某种hash值)应该路由到哪个partition上。分区策略可以根据实际需求来，为了某些特殊的需求，甚至可以将所有数据只写到一个partition上。
- 与生产者类似，从哪个partition上消费消息，是由消费者客户端(consumer client)指定的。换言之，要想获取到topic的所有消息，必须同时和所有partition位于的broker建立链接来消费消息。
- 无论生产还是消费消息，必须从partition的Leader上进行。否则服务器将返回NotLeaderForPartition错误。所以在生产或消费消息时，要先发送Topic Metadata Request，服务器会返回topic都有多少分区，哪个broker是这些分区的Leader，以及这些broker主机的地址和端口信息。
Batching
- kafka支持批处理请求，所以尽量将小的请求批量处理以提高效率。一个实际生产环境遇到的场景：服务器上有约1000个topic，每个topic平均3个partition。需求是要获取所有topic partition的当前offset。如果为每个partition发送一次api请求，那就是3000多次请求，一共耗时2分钟左右。改为一次请求获取所有partition的offset后，耗时不到2秒。
Versioning
- 所有api都有一个版本号参数，版本号从0开始。需要使用与服务器对应的版本api请求，如果发送服务器不支持的版本，则请求会被拒绝。
Protocol Primitive Types
协议中的请求和响应都由下面3种类型的数据组成。
- 定长基本类型： BOOLEAN, INT8, INT16, INT32, INT64, UINT32, VARINT, VARLONG, UUID, FLOAT64
- 变长基本类型：STRING, BYTES。STRING由一个表示长度的数字N(int16)和N个字符组成。BYTES类似，也是由一个表示长度的数字N(int32)和N个字符组成。2者都有COMPACT、NULLABLE、COMPACT_NULLABLE3个变种。
- 数组：由表示个数的数字N(int32)和N个重复结构体组成，单个结构体是上面2种基本类型的组合。
Request & Response
请求和响应的前4个字节都是一个表示请求体或响应体多大的数字:
RequestOrResponse => Size (RequestMessage | ResponseMessage)
message_size => int32
RequestMessage ｜ ResponseMessage
| Field | 长度 | 含义 |
|---|---|---|
| message_size | 4字节 | message_size给出了后续请求或响应消息的字节(bytes)长度。客户端可以先读取4字节的长度N，然后读取并解析后续的N字节请求内容。 |
注意：后面介绍中将不会再专门列出message_size域，默认在头部包含该域值
Request Header
所有请求都具有相同格式的请求头
RequestMessage => api_key api_version correlation_id client_id
api_key => int16
api_version => int16
correlation_id => int32
client_id => string
MetadataRequest | ProduceRequest | FetchRequest | ListOffsetRequest |......
| Field | 长度 | 描述 |
|---|---|---|
| api_key | 2字节 | API Code |
| api_version | 2字节 | API 版本号 |
| correlation_id | 4字节 | 自定义消息id (response中会将该值原样返回) |
| client_id | 2字节+N | 自定义ClientId |
Response Header
所有响应都具有相同格式的响应头
ResponseMessage => correlation_id
correlation_id => int32
MetadataResponse | ProduceResponse | FetchResponse | ListOffsetResponse |......
| Field | 长度 | 描述 |
|---|---|---|
| correlation_id | 4字节 | 自定义消息id(request中的correlation_id) |
