Kafka消息生產(chǎn)及消費大體流程

發(fā)送流程

image.png

1、名詞含義：

1）Producer ：消息生產(chǎn)者佳晶，就是向kafka broker發(fā)消息的客戶端宠叼。

2）Consumer ：消息消費者，向kafka broker取消息的客戶端

3）Topic ：可以理解為一個隊列作儿。

4） Consumer Group （CG）：這是kafka用來實現(xiàn)一個topic消息的廣播（發(fā)給所有的consumer）和單播（發(fā)給任意一個consumer）的手段强胰。一個topic可以有多個CG尚镰。topic的消息會復制（不是真的復制，是概念上的）到所有的CG哪廓，但每個partion只會把消息發(fā)給該CG中的一個consumer。如果需要實現(xiàn)廣播初烘，只要每個consumer有一個獨立的CG就可以了涡真。要實現(xiàn)單播只要所有的consumer在同一個CG分俯。用CG還可以將consumer進行自由的分組而不需要多次發(fā)送消息到不同的topic。

5）Broker ：一臺kafka服務器就是一個broker哆料。一個集群由多個broker組成缸剪。一個broker可以容納多個topic。

6）Partition：為了實現(xiàn)擴展性东亦，一個非常大的topic可以分布到多個broker（即服務器）上杏节，一個topic可以分為多個partition，每個partition是一個有序的隊列典阵。partition中的每條消息都會被分配一個有序的id（offset）奋渔。kafka只保證按一個partition中的順序?qū)⑾l(fā)給consumer，不保證一個topic的整體（多個partition間）的順序壮啊。

7）Offset：kafka的存儲文件都是按照offset.kafka來命名嫉鲸，用offset做名字的好處是方便查找。例如你想找位于2049的位置歹啼，只要找到2048.kafka的文件即可玄渗。當然the first offset就是00000000000.kafka

2、當要生產(chǎn)一條消息時大致流程如下：

1）Producer 會計算本條消息需要發(fā)送的partition狸眼。

2）Producer 根據(jù)發(fā)送的分區(qū)藤树，向zookeeper獲取對應partition的leader信息，發(fā)送消息到leader所在的broker拓萌。

3）leader在本地記錄該消息岁钓。

4）follower通過輪詢監(jiān)控到leader新寫入消息，主動拉取消息司志。

5）follower同步消息成功向leader發(fā)送ack甜紫。

6）leader收到所有follower同步的消息，向producer發(fā)送確認ack骂远。

3囚霸、詳細流程

3.1、首先是我們的producter調(diào)用kafka提供的rpc接口的send方法將我們要發(fā)送的消息給到kafka服務器激才，這其中會產(chǎn)生一個問題拓型，即當生產(chǎn)一條消息時(某個Topic)，它只會屬于某一個分片瘸恼，那么這個消息應該歸屬哪一臺服務器劣挫，或者哪一個分片呢？ procducer怎么知道該發(fā)送到哪個機器上东帅？ 通常這類問題有兩個解決方案：

1）有proxy專門負責這類判定压固、轉發(fā)，client對細節(jié)無感知(類似MySQL中間件代理)

2）Producer(client)端掌握server端的詳細信息靠闭，實現(xiàn)重型Client(類似redis cluster)

而kafka使用的就是第二種方案帐我，Producer 可以通過bootstrap.servers中任意一個kafka實例坎炼，拉取到所有元信息，比如某個Topic有多少個Partition拦键，每個Partition的leader的地址谣光，這些元信息Producer會定時輪詢更新。

每個kafka節(jié)點都有完整的元信息芬为，Producer可以通過任意節(jié)點拉取萄金，源頭維護于Zookeeper之中，當集群中的Partition等元信息發(fā)生變更媚朦，Controller節(jié)點會逐一推送給其他Broker最新信息氧敢。zookeeper的作用其實主要是兩個，一是作為存儲莲镣，二是基于其Watch能力做事件驅(qū)動(例如元信息更新推送)福稳。

3.2、消息到達kafka服務器后瑞侮，會先經(jīng)過攔截器的圆，接著進入序列化器。序列化器主要用于對消息的Key和Value進行序列化半火。接著進入分區(qū)器選擇消息的分區(qū)越妈。

3.3、上面這幾步完成之后钮糖，消息會進入到一個名為RecordAccumulator的緩沖隊列梅掠，這個隊列默認32M。當滿足以下兩個條件的任意一個之后店归，消息由sender線程發(fā)送阎抒。

條件一：消息累計達到batch.size，默認是16kb消痛。

條件二：等待時間達到linger.ms且叁，默認是0毫秒。

所以在默認情況下秩伞，由于等待時間是0毫秒逞带，所以只要消息來一條就會發(fā)送一條。

3.4纱新、Sender線程首先會通過sender讀取數(shù)據(jù)展氓，并創(chuàng)建發(fā)送的請求，針對Kafka集群里的每一個Broker脸爱，都會有一個InFlightRequests請求隊列存放在NetWorkClient中遇汞，默認每個InFlightRequests請求隊列中緩存5個請求。接著這些請求就會通過Selector發(fā)送到Kafka集群中。

3.5勺疼、當請求發(fā)送到Kafka集群后教寂，Kafka集群會返回對應的acks信息。生產(chǎn)者可以根據(jù)具體的情況選擇處理acks信息执庐。

0：生產(chǎn)者發(fā)送過來的數(shù)據(jù)，不需要等數(shù)據(jù)落盤應答导梆。

1：生產(chǎn)者發(fā)送過來的數(shù)據(jù)轨淌，Leader收到數(shù)據(jù)后應答。

-1（all）：生產(chǎn)者發(fā)送過來的數(shù)據(jù)看尼，Leader+和isr隊列里面的所有節(jié)點收齊數(shù)據(jù)后應答递鹉。默認值是-1，-1和all是等價的藏斩。

到此躏结，一條消息的發(fā)送任務就結束了。

消費流程

Kafka的消費方式

消費方式常見會有兩種狰域，即推送模型（push）和拉取模型（pull）媳拴。

Push方式在將消息推送到消費者后，就會將這條消息標記為 已消費 狀態(tài)兆览，但是如果在消費者消費消息時出現(xiàn)異城龋或者其他意外情況導致消息其實并沒有被正常消費，那么該條消息就有可能存在丟失的情況抬探。需要解決這個問題子巾，就需要在kafka服務器在推送消息后不能直接標記消息為 已消費 ，而是會存在一個中間態(tài)小压，即 已發(fā)送 线梗，待消費者正常消費完成后再將狀態(tài)改成 已消費，這種方式帶來了更多的額外消耗怠益。

Poll方式則是由消費者自行記錄消息消費狀態(tài)仪搔，每個消費者獨立且有序的拉取每個分區(qū)的消息到本地。

Kafka使用的就是poll方式溉痢。

image.png

消息拉取執(zhí)行流程：

1僻造、Consumer從 zookeeper當中獲取到 partition 以及 consumer 對應的 offset （默認從zookeeper中獲取上一次消費的offset）

2、找到該分區(qū)的leader孩饼，拉取數(shù)據(jù)

3髓削、leader從本地log（日志）當中讀取數(shù)據(jù)，最終返回給消費者

4镀娶、最終拉取完數(shù)據(jù)立膛，提交offset給zookeeper。

kafka實際應用場景

1. 消息

kafka更好的替換傳統(tǒng)的消息系統(tǒng)，消息系統(tǒng)被用于各種場景宝泵，與大多數(shù)消息系統(tǒng)比較kafka有更好的吞吐量內(nèi)置分區(qū)好啰，副本和故障轉移，這有利于處理大規(guī)模的消息儿奶。

多用于解耦消息的發(fā)送方和消費方框往。

根據(jù)我們的經(jīng)驗消息往往用于較低的吞吐量，但需要低的端到端延遲并需要提供強大的耐用性的保證闯捎。在這一領域的kafka比得上傳統(tǒng)的消息系統(tǒng)椰弊，如ActiveMQ或RabbitMQ等。

2. 網(wǎng)站活動追蹤

kafka原本的使用場景是用戶的活動追蹤瓤鼻，網(wǎng)站的活動（網(wǎng)頁游覽秉版，搜索或其他用戶的操作信息）發(fā)布到不同的話題中心，這些消息可實時處理實時監(jiān)測也可加載到Hadoop或離線處理數(shù)據(jù)倉庫茬祷。

3. 指標

kafka也經(jīng)常用來記錄運營監(jiān)控數(shù)據(jù)清焕。包括收集各種分布式應用的數(shù)據(jù)，生產(chǎn)各種操作的集中反饋祭犯，比如報警和報告秸妥，用于監(jiān)測數(shù)據(jù)，分布式應用程序生成的統(tǒng)計數(shù)據(jù)集中聚合盹憎。

4. 日志聚合

許多人使用Kafka作為日志聚合解決方案的替代品筛峭。日志聚合通常從服務器中收集物理日志文件，并將它們放在中央位置（可能是文件服務器或HDFS）進行處理陪每。Kafka抽象出文件的細節(jié)影晓，并將日志或事件數(shù)據(jù)更清晰地抽象為消息流。這允許更低延遲的處理并更容易支持多個數(shù)據(jù)源和分布式數(shù)據(jù)消費檩禾。

5. 流處理

kafka中消息處理一般包含多個階段挂签。其中原始輸入數(shù)據(jù)是從kafka主題消費的，然后匯總盼产，豐富饵婆，或者以其他的方式處理轉化為新主題，例如戏售，一個推薦新聞文章侨核，文章內(nèi)容可能從“articles”主題獲取灌灾；然后進一步處理內(nèi)容搓译，得到一個處理后的新內(nèi)容，最后推薦給用戶锋喜。這種處理是基于單個主題的實時數(shù)據(jù)流些己。

除了Kafka Streams還有ApacheStorm和Apache Samza可選擇豌鸡。

6. 事件采集

事件采集是一種應用程序的設計風格，其中狀態(tài)的變化根據(jù)時間的順序記錄下來段标，kafka支持這種非常大的存儲日志數(shù)據(jù)的場景涯冠。

7. 提交日志

kafka可以作為一種分布式的外部日志，可幫助節(jié)點之間復制數(shù)據(jù)逼庞，并作為失敗的節(jié)點來恢復數(shù)據(jù)重新同步蛇更，kafka的日志壓縮功能很好的支持這種用法，這種用法類似于Apacha BookKeeper項目赛糟。

8. 大數(shù)據(jù)的實時計算

kafka被應用到大數(shù)據(jù)處理械荷，如與spark、storm等整合虑灰。