Kafka入門教程與詳解

1?Kafka入門教程

1.1?消息隊列（Message Queue)

Message Queue消息傳送系統提供傳送服務。消息傳送依賴于大量支持組件，這些組件負責處理連接服務、消息的路由和傳送、持久性、安全性以及日志記錄。消息服務器可以使用一個或多個代理實例。

JMS（Java Messaging Service）是Java平臺上有關面向消息中間件(MOM)的技術規范，它便于消息系統中的Java應用程序進行消息交換,并且通過提供標準的產生、發送、接收消息的接口簡化企業應用的開發，翻譯為Java消息服務。

1.2?MQ消息模型

KafkaMQ消息模型圖1-1

1.3?MQ消息隊列分類

消息隊列分類：點對點和發布/訂閱兩種：

1、點對點：

消息生產者生產消息發送到queue中，然后消息消費者從queue中取出并且消費消息。

消息被消費以后，queue中不再有存儲，所以消息消費者不可能消費到已經被消費的消息。Queue支持存在多個消費者，但是對一個消息而言，只會有一個消費者可以消費。

2、發布/訂閱：

消息生產者（發布）將消息發布到topic中，同時有多個消息消費者（訂閱）消費該消息。和點對點方式不同，發布到topic的消息會被所有訂閱者消費。

1.4?MQ消息隊列對比

1、RabbitMQ：支持的協議多，非常重量級消息隊列，對路由(Routing)，負載均衡(Loadbalance)或者數據持久化都有很好的支持。

2、ZeroMQ：號稱最快的消息隊列系統，尤其針對大吞吐量的需求場景，擅長的高級/復雜的隊列，但是技術也復雜，并且只提供非持久性的隊列。

3、ActiveMQ：Apache下的一個子項，類似ZeroMQ，能夠以代理人和點對點的技術實現隊列。

4、Redis：是一個key-Value的NOSql數據庫，但也支持MQ功能，數據量較小，性能優于RabbitMQ，數據超過10K就慢的無法忍受。

1.5?Kafka簡介

Kafka是分布式發布-訂閱消息系統,它最初由?LinkedIn?公司開發，使用?Scala語言編寫,之后成為?Apache?項目的一部分。在Kafka集群中，沒有“中心主節點”的概念，集群中所有的服務器都是對等的，因此，可以在不做任何配置的更改的情況下實現服務器的的添加與刪除，同樣的消息的生產者和消費者也能夠做到隨意重啟和機器的上下線。

Kafka消息系統生產者和消費者部署關系圖1-2

Kafka消息系統架構圖1-3

1.6?Kafka術語介紹

1、消息生產者：即：Producer，是消息的產生的源頭，負責生成消息并發送到Kafka

服務器上。

2、消息消費者：即：Consumer，是消息的使用方，負責消費Kafka服務器上的消息。

3、主題：即：Topic，由用戶定義并配置在Kafka服務器，用于建立生產者和消息者之間的訂閱關系：生產者發送消息到指定的Topic下，消息者從這個Topic下消費消息。

4、消息分區：即：Partition，一個Topic下面會分為很多分區，例如：“kafka-test”這個Topic下可以分為6個分區，分別由兩臺服務器提供，那么通常可以配置為讓每臺服務器提供3個分區，假如服務器ID分別為0、1，則所有的分區為0-0、0-1、0-2和1-0、1-1、1-2。Topic物理上的分組，一個?topic可以分為多個?partition，每個?partition?是一個有序的隊列。partition中的每條消息都會被分配一個有序的?id（offset）。

5、Broker：即Kafka的服務器，用戶存儲消息，Kafa集群中的一臺或多臺服務器統稱為?broker。

6、消費者分組：Group，用于歸組同類消費者，在Kafka中，多個消費者可以共同消息一個Topic下的消息，每個消費者消費其中的部分消息，這些消費者就組成了一個分組，擁有同一個分組名稱，通常也被稱為消費者集群。

7、Offset：消息存儲在Kafka的Broker上，消費者拉取消息數據的過程中需要知道消息在文件中的偏移量，這個偏移量就是所謂的Offset。

1.7?Kafka中Broker

1、Broker：即Kafka的服務器，用戶存儲消息，Kafa集群中的一臺或多臺服務器統稱為?broker。

2、Message在Broker中通Log追加的方式進行持久化存儲。并進行分區（patitions)。

3、為了減少磁盤寫入的次數,broker會將消息暫時buffer起來,當消息的個數(或尺寸)達到一定閥值時,再flush到磁盤,這樣減少了磁盤IO調用的次數。

4、Broker沒有副本機制，一旦broker宕機，該broker的消息將都不可用。Message消息是有多份的。

5、Broker不保存訂閱者的狀態，由訂閱者自己保存。

6、無狀態導致消息的刪除成為難題（可能刪除的消息正在被訂閱），kafka采用基于時間的SLA(服務水平保證)，消息保存一定時間（通常為7天）后會被刪除。

7、消息訂閱者可以rewind back到任意位置重新進行消費，當訂閱者故障時，可以選擇最小的offset(id)進行重新讀取消費消息。

1.8?Kafka的Message組成

1、Message消息：是通信的基本單位，每個?producer?可以向一個?topic（主題）發布一些消息。

2、Kafka中的Message是以topic為基本單位組織的，不同的topic之間是相互獨立的。每個topic又可以分成幾個不同的partition(每個topic有幾個partition是在創建topic時指定的)，每個partition存儲一部分Message。

3、partition中的每條Message包含了以下三個屬性：

offset ?????即：消息唯一標識:對應類型：long

MessageSize?對應類型：int32

data????? ??是message的具體內容。

1.9?Kafka的Partitions分區

1、Kafka基于文件存儲.通過分區，可以將日志內容分散到多個server上,來避免文件尺寸達到單機磁盤的上限，每個partiton都會被當前server(kafka實例)保存。

2、可以將一個topic切分多任意多個partitions，來消息保存/消費的效率。

3、越多的partitions意味著可以容納更多的consumer，有效提升并發消費的能力。

1.10?Kafka的Consumers

1、消息和數據消費者，訂閱?topics并處理其發布的消息的過程叫做?consumers。

2、在?kafka中,我們可以認為一個group是一個“訂閱者”，一個Topic中的每個partions，只會被一個“訂閱者”中的一個consumer消費，不過一個?consumer可以消費多個partitions中的消息（消費者數據小于Partions的數量時）。注意：kafka的設計原理決定，對于一個topic，同一個group中不能有多于partitions個數的consumer同時消費，否則將意味著某些consumer將無法得到消息。

3、一個partition中的消息只會被group中的一個consumer消息。每個group中consumer消息消費互相獨立。

1.11?Kafka的持久化

1、一個Topic可以認為是一類消息，每個topic將被分成多partition(區),每個partition在存儲層面是append log文件。任何發布到此partition的消息都會被直接追加到log文件的尾部，每條消息在文件中的位置稱為offset（偏移量），partition是以文件的形式存儲在文件系統中。

2、Logs文件根據broker中的配置要求,保留一定時間后刪除來釋放磁盤空間。

Kafka消息分區Partition圖1-4

Partition：

Topic物理上的分組，一個?topic可以分為多個?partition，每個?partition?是一個有序的隊列。partition中的每條消息都會被分配一個有序的?id（offset）。

3、為數據文件建索引：稀疏存儲，每隔一定字節的數據建立一條索引。下圖為一個partition的索引示意圖：

Kafka消息分區Partition索引圖1-5

1.12?Kafka的分布式實現：

Kafka分布式關系圖1-6

Kafka生產環境關系圖1-7

1.13?Kafka的通訊協議：

1、Kafka的Producer、Broker和Consumer之間采用的是一套自行設計基于TCP層的協議，根據業務需求定制，而非實現一套類似ProtocolBuffer的通用協議。

2、基本數據類型：（Kafka是基于Scala語言實現的，類型也是Scala中的數據類型）

定長數據類型：int8,int16,int32和int64，對應到Java中就是byte, short, int和long。

變長數據類型：bytes和string。變長的數據類型由兩部分組成，分別是一個有符號整數N(表示內容的長度)和N個字節的內容。其中，N為-1表示內容為null。bytes的長度由int32表示，string的長度由int16表示。

數組：數組由兩部分組成，分別是一個由int32類型的數字表示的數組長度N和N個元素。

3、Kafka通訊的基本單位是Request/Response。

4、基本結構：

RequestOrResponse => MessageSize(RequestMessage | ResponseMessage)

名稱

類型

描術

MessageSize

int32

表示RequestMessage或者ResponseMessage的長度

RequestMessage

ResponseMessage

—

5、通訊過程：

客戶端打開與服務器端的Socket

往Socket寫入一個int32的數字(數字表示這次發送的Request有多少字節)

服務器端先讀出一個int32的整數從而獲取這次Request的大小

然后讀取對應字節數的數據從而得到Request的具體內容

服務器端處理了請求后，也用同樣的方式來發送響應。

6、RequestMessage結構：

RequestMessage => ApiKey ApiVersionCorrelationId ClientId Request

名稱

類型

描術

ApiKey

int16

表示這次請求的API編號

ApiVersion

int16

表示請求的API的版本，有了版本后就可以做到后向兼容

CorrelationId

int32

由客戶端指定的一個數字唯一標示這次請求的id，服務器端在處理完請求后也會把同樣的CorrelationId寫到Response中，這樣客戶端就能把某個請求和響應對應起來了。

ClientId

string

客戶端指定的用來描述客戶端的字符串，會被用來記錄日志和監控，它唯一標示一個客戶端。

Request

—

Request的具體內容。

7、ResponseMessage結構：

ResponseMessage => CorrelationId Response

名稱

類型

描術

CorrelationId

int32

對應Request的CorrelationId。

Response

—

對應Request的Response，不同的Request的Response的字段是不一樣的。

Kafka采用是經典的Reactor(同步IO)模式，也就是1個Acceptor響應客戶端的連接請求，N個Processor來讀取數據，這種模式可以構建出高性能的服務器。

8、Message結構：

Message:Producer生產的消息,鍵-值對

Message => Crc MagicByte Attributes KeyValue

名稱

類型

描術

CRC

int32

表示這條消息(不包括CRC字段本身)的校驗碼。

MagicByte

int8

表示消息格式的版本，用來做后向兼容，目前值為0。

Attributes

int8

表示這條消息的元數據，目前最低兩位用來表示壓縮格式。

Key

bytes

表示這條消息的Key，可以為null。

Value

bytes

表示這條消息的Value。Kafka支持消息嵌套，也就是把一條消息作為Value放到另外一條消息里面。

9、MessageSet結構：

MessageSet:用來組合多條Message，它在每條Message的基礎上加上了Offset和MessageSize

MessageSet => [Offset MessageSize Message]

名稱

類型

描術

Offset

int64

它用來作為log中的序列號，Producer在生產消息的時候還不知道具體的值是什么，可以隨便填個數字進去。

MessageSize

int32

表示這條Message的大小。

Message

-

表示這條Message的具體內容，其格式見上一小節。

10、?????Request/Respone和Message/MessageSet的關系：

Request/Response是通訊層的結構，和網絡的7層模型對比的話，它類似于TCP層。

Message/MessageSet定義的是業務層的結構，類似于網絡7層模型中的HTTP層。Message/MessageSet只是Request/Response的payload中的一種數據結構。

備注：Kafka的通訊協議中不含Schema，格式也比較簡單，這樣設計的好處是協議自身的Overhead小，再加上把多條Message放在一起做壓縮，提高壓縮比率，從而在網絡上傳輸的數據量會少一些。

1.14?數據傳輸的事務定義：

1、at most once:最多一次,這個和JMS中”非持久化”消息類似.發送一次，無論成敗，將不會重發。

at most once:消費者fetch消息,然后保存offset，然后處理消息;當client保存offset之后，但是在消息處理過程中出現了異常，導致部分消息未能繼續處理.那么此后”未處理”的消息將不能被fetch到，這就是“atmost once”。

2、at least once:消息至少發送一次，如果消息未能接受成功，可能會重發，直到接收成功。

at least once:消費者fetch消息，然后處理消息，然后保存offset.如果消息處理成功之后，但是在保存offset階段zookeeper異常導致保存操作未能執行成功，這就導致接下來再次fetch時可能獲得上次已經處理過的消息，這就是“atleast once”，原因offset沒有及時的提交給zookeeper，zookeeper恢復正常還是之前offset狀態。

3、exactly once:消息只會發送一次。

exactly once: kafka中并沒有嚴格的去實現(基于2階段提交，事務)，我們認為這種策略在kafka中是沒有必要的。

注：通常情況下“at-least-once”是我們首選。(相比at most once而言，重復接收數據總比丟失數據要好)。

二、消息隊列之Kafka工作原理與安裝介紹

2.1消息隊列之Kafka工作原理 -- broker

2.2消息隊列之Kafka工作原理 -- topic

2.3消息隊列之Kafka工作原理 – partition

2.4消息隊列之Kafka安裝介紹

版本

Apache Kafka 與 Confluent Platform

Docker鏡像 Confluent kafka 的docker鏡像

客戶端工具

Apache Kafka的Python客戶端：kafka-python

Confluent kafka的Python客戶端： confluent-kafka-python

git地址

使用文檔

2.5消息隊列之Kafka使用介紹

Kafka啟動：

單節點單broker 單節點多broker

Kafka使用時的顯著特征

分區之間是無序的，但分區內的消息是有序的

對于topic的消費，消費者的數量 應 不多于 該topic分區的數量，否則多余的消費者將必定無法接收到消息

一個消費者可同時消費多個topic

在訂閱消費時，Kafka保證每條消息在同一個Consumer Group里只會被某一個Consumer消費

總結：掌握原理 活用文檔 多實踐

Kafka 數據結構

版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。