Kafka的通訊協議
標簽:kafka
Kafka的Producer、Broker和Consumer之間采用的是一套自行設計的基於TCP層的協議。Kafka的這套協議完全是為了Kafka自身的業務需求而定制的，而非要實現一套類似於Protocol Buffer的通用協議。本文將介紹這套協議的相關內容。
基本數據類型
- 定長數據類型：int8,int16,int32和int64，對應到Java中就是byte, short, int和long。
- 變長數據類型：bytes和string。變長的數據類型由兩部分組成，分別是一個有符號整數N(表示內容的長度)和N個字節的內容。其中，N為-1表示內容為null。bytes的長度由int32表示，string的長度由int16表示。
- 數組：數組由兩部分組成，分別是一個由int32類型的數字表示的數組長度N和N個元素。
Request和Response的基本結構
Kafka中兩個角色之間通訊的基本單位是Request/Response，Request和Response的基本結構如下：
RequestOrResponse => MessageSize (RequestMessage | ResponseMessage)
其中各字段的含義為：
| 名稱 | 類型 | 描述 |
|---|---|---|
| MessageSize | int32 | 表示RequestMessage或者ResponseMessage的長度 |
| RequestMessage/ResponseMessage | - | 表示Request或者Response的內容，在下面將會介紹其具體格式。 |
這個結構定義了通訊雙方交換數據的基本結構。通訊的過程可以簡單地表示為：客戶端打開與服務器端的Socket，然后往Socket寫入一個int32的數字表示這次發送的Request有多少字節，然后繼續往Socket中寫入對應字節數的數據。服務器端先讀出一個int32的整數從而獲取這次Request的大小，然后讀取對應字節數的數據從而得到Request的具體內容。服務器端處理了請求后，也用同樣的方式來發送響應。
RequestMessage的結構
RequestMessage的結構如下：
RequestMessage => ApiKey ApiVersion CorrelationId ClientId Request
| 名稱 | 類型 | 描述 |
|---|---|---|
| ApiKey | int16 | 表示這次請求的API編號 |
| ApiVersion | int16 | 表示請求的API的版本，有了版本后就可以做到后向兼容 |
| CorrelationId | int32 | 由客戶端指定的一個數字唯一標示這次請求的id，服務器端在處理完請求后也會把同樣的CorrelationId寫到Response中，這樣客戶端就能把某個請求和響應對應起來了。 |
| ClientId | string | 客戶端指定的用來描述客戶端的字符串，會被用來記錄日志和監控，它唯一標示一個客戶端。 |
| Request | - | Request的具體內容。 |
ResponseMessage的結構
ResponseMessage的結構如下：
ResponseMessage => CorrelationId Response
| 名稱 | 類型 | 描述 |
|---|---|---|
| CorrelationId | int32 | 對應Request的CorrelationId。 |
| Response | - | 對應Request的Response，不同的Request的Response的字段是不一樣的。 |
Message
Kafka是一個分布式消息系統，Producer生產消息並推送(Push)給Broker，然后Consumer再從Broker那里取走(Pull)消息。Producer生產的消息就是由Message來表示的，對用戶來講，它就是鍵-值對，來看看它的結構。
Message => Crc MagicByte Attributes Key Value
| 名稱 | 類型 | 描述 |
|---|---|---|
| CRC | int32 | 表示這條消息(不包括CRC字段本身)的校驗碼 |
| MagicByte | int8 | 表示消息格式的版本，用來做后向兼容，目前值為0 |
| Attributes | int8 | 表示這條消息的元數據，目前最低兩位用來表示壓縮格式 |
| Key | bytes | 表示這條消息的Key，可以為null |
| Value | bytes | 表示這條消息的Value。Kafka支持消息嵌套，也就是把一條消息作為Value放到另外一條消息里面。 |
MessageSet
MessageSet用來組合多條Message，它在每條Message的基礎上加上了Offset和MessageSize，其結構是：
MessageSet => [Offset MessageSize Message]
它的含義是MessageSet是個數組，數組的每個元素由三部分組成，分別是Offset，MessageSize和Message，它們的含義分別是：
| 名稱 | 類型 | 描述 |
|---|---|---|
| Offset | int64 | 它用來作為log中的序列號，Producer在生產消息的時候還不知道具體的值是什么，可以隨便填個數字進去 |
| MessageSize | int32 | 表示這條Message的大小 |
| Message | - | 表示這條Message的具體內容，其格式見上一小節。 |
Message的壓縮
Kafka支持下面幾種壓縮方式，
| 壓縮方式 | 編碼 |
|---|---|
| 不壓縮 | 0 |
| Gzip | 1 |
| Snappy | 2 |
| LZ4 | 3 |
其中編碼就是Message的Attribute的最低兩位的值。
因為單條消息中重復內容可能不多，所以通常把多條消息放在一起組成MessageSet，然后再把MessageSet放到一條Message里面去，從而提高壓縮比率。
Request/Respone和Message/MessageSet的關系
- Request/Response是通訊層的結構，和網絡的7層模型對比的話，它類似於TCP層。
- Message/MessageSet定義的是業務層的結構，類似於網絡7層模型中的HTTP層。Message/MessageSet只是Request/Response的payload中的一種數據結構。
小結
Kafka的通訊協議中不含Schema，格式也比較簡單，這樣設計的好處是協議自身的Overhead小，再加上把多條Message放在一起做壓縮，提高壓縮比率，從而在網絡上傳輸的數據量會少一些。