Kafka入門

為什么要用消息中間件？

異步處理

場景說明：用戶注冊后，需要發(fā)注冊郵件和注冊短信。傳統的做法有兩種1.串行的方式；2.并行方式。

串行方式：將注冊信息寫入數據庫成功后，發(fā)送注冊郵件，再發(fā)送注冊短信。以上三個任務全部完成后，返回給客戶端。

[外鏈圖片轉存失敗,源站可能有防盜鏈機制,建議將圖片保存下來直接上傳(img-QSbXkisA-1686193157416)(file:///C:\Users\ADMINI~1\AppData\Local\Temp\msohtmlclip1\01\clip_image002.jpg)]

（2）并行方式：將注冊信息寫入數據庫成功后，發(fā)送注冊郵件的同時，發(fā)送注冊短信。以上三個任務完成后，返回給客戶端。與串行的差別是，并行的方式可以提高處理的時間。

假設三個業(yè)務節(jié)點每個使用50毫秒鐘，不考慮網絡等其他開銷，則串行方式的時間是150毫秒，并行的時間可能是100毫秒。

[外鏈圖片轉存失敗,源站可能有防盜鏈機制,建議將圖片保存下來直接上傳(img-9p13A393-1686193157418)(file:///C:\Users\ADMINI~1\AppData\Local\Temp\msohtmlclip1\01\clip_image004.jpg)]

小結：如以上案例描述，傳統的方式系統的性能（并發(fā)量，吞吐量，響應時間）會有瓶頸。如何解決這個問題呢？

引入消息隊列，將不是必須的業(yè)務邏輯，異步處理。

[外鏈圖片轉存失敗,源站可能有防盜鏈機制,建議將圖片保存下來直接上傳(img-8RExvNKZ-1686193157419)(file:///C:\Users\ADMINI~1\AppData\Local\Temp\msohtmlclip1\01\clip_image006.jpg)]

按照以上約定，用戶的響應時間相當于是注冊信息寫入數據庫的時間，也就是50毫秒。注冊郵件，發(fā)送短信寫入消息隊列后，直接返回，因此寫入消息隊列的速度很快，基本可以忽略，因此用戶的響應時間可能是50毫秒。因此架構改變后，系統的吞吐量提高到每秒20 QPS。比串行提高了3倍，比并行提高了兩倍。

應用解耦

場景說明：用戶下單后，訂單系統需要通知庫存系統。傳統的做法是，訂單系統調用庫存系統的接口。

傳統模式的缺點：

1） 假如庫存系統無法訪問，則訂單減庫存將失敗，從而導致訂單失敗；

2） 訂單系統與庫存系統耦合；

[外鏈圖片轉存失敗,源站可能有防盜鏈機制,建議將圖片保存下來直接上傳(img-CZbRjNwt-1686193157420)(file:///C:\Users\ADMINI~1\AppData\Local\Temp\msohtmlclip1\01\clip_image008.jpg)]

如何解決以上問題呢？引入應用消息隊列后的方案

訂單系統：用戶下單后，訂單系統完成持久化處理，將消息寫入消息隊列，返回用戶訂單下單成功。

庫存系統：訂閱下單的消息，采用拉/推的方式，獲取下單信息，庫存系統根據下單信息，進行庫存操作。

[外鏈圖片轉存失敗,源站可能有防盜鏈機制,建議將圖片保存下來直接上傳(img-c7GtV11T-1686193157422)(file:///C:\Users\ADMINI~1\AppData\Local\Temp\msohtmlclip1\01\clip_image010.jpg)]

假如：在下單時庫存系統不能正常使用。也不影響正常下單，因為下單后，訂單系統寫入消息隊列就不再關心其他的后續(xù)操作了。實現訂單系統與庫存系統的應用解耦。

流量削峰

流量削峰也是消息隊列中的常用場景，一般在秒殺或團搶活動中使用廣泛。

應用場景：秒殺活動，一般會因為流量過大，導致流量暴增，應用掛掉。為解決這個問題，一般需要在應用前端加入消息隊列：可以控制活動的人數；可以緩解短時間內高流量壓垮應用。

[外鏈圖片轉存失敗,源站可能有防盜鏈機制,建議將圖片保存下來直接上傳(img-KkntO33S-1686193157423)(file:///C:\Users\ADMINI~1\AppData\Local\Temp\msohtmlclip1\01\clip_image012.gif)]

[外鏈圖片轉存失敗,源站可能有防盜鏈機制,建議將圖片保存下來直接上傳(img-8cQLq6b0-1686193157425)(file:///C:\Users\ADMINI~1\AppData\Local\Temp\msohtmlclip1\01\clip_image014.gif)]

用戶的請求，服務器接收后，首先寫入消息隊列。假如消息隊列長度超過最大數量，則直接拋棄用戶請求或跳轉到錯誤頁面；秒殺業(yè)務根據消息隊列中的請求信息，再做后續(xù)處理。

日志處理

日志處理是指將消息隊列用在日志處理中，比如Kafka的應用，解決大量日志傳輸的問題。架構簡化如下：

日志采集客戶端，負責日志數據采集，定時寫入Kafka隊列：Kafka消息隊列，負責日志數據的接收，存儲和轉發(fā)；日志處理應用：訂閱并消費kafka隊列中的日志數據；

為什么選擇Kafka？

消息中間件的編年史

Kafka的外在表現和內在設計

kafka最初是LinkedIn的一個內部基礎設施系統。最初開發(fā)的起因是，LinkedIn雖然有了數據庫和其他系統可以用來存儲數據，但是缺乏一個可以幫助處理持續(xù)數據流的組件。

所以在設計理念上，開發(fā)者不想只是開發(fā)一個能夠存儲數據的系統，如關系數據庫、Nosql數據庫、搜索引擎等等，更希望把數據看成一個持續(xù)變化和不斷增長的流，并基于這樣的想法構建出一個數據系統，一個數據架構。

Kafka外在表現很像消息系統，允許發(fā)布和訂閱消息流，但是它和傳統的消息系統有很大的差異：

1、Kafka是個現代分布式系統，以集群的方式運行，可以自由伸縮。

2、Kafka可以按照要求存儲數據，保存多久都可以，

3、流式處理將數據處理的層次提示到了新高度，消息系統只會傳遞數據，Kafka的流式處理能力可以讓我們用很少的代碼就能動態(tài)地處理派生流和數據集。所以Kafka不僅僅是個消息中間件。

Kafka不僅僅是一個消息中間件，同時它是一個流平臺，這個平臺上可以發(fā)布和訂閱數據流（Kafka的流，有一個單獨的包Stream的處理），并把他們保存起來，進行處理，這個是Kafka作者的設計理念。

大數據領域，Kafka還可以看成實時版的Hadoop，但是還是有些區(qū)別，Hadoop可以存儲和定期處理大量的數據文件，往往以TB計數，而Kafka可以存儲和持續(xù)處理大型的數據流。Hadoop主要用在數據分析上，而Kafka因為低延遲，更適合于核心的業(yè)務應用上。

Kafka名字的由來：卡夫卡與法國作家馬塞爾·普魯斯特，愛爾蘭作家詹姆斯·喬伊斯并稱為西方現代主義文學的先驅和大師。《變形記》是卡夫卡的短篇代表作，是卡夫卡的藝術成就中的一座高峰，被認為是20世紀最偉大的小說作品之一（達到管理層的高度同學可以多看下人文相關的書籍，增長管理知識和人格魅力）。

本次課程，將會以kafka_2.13-3.3.1版本做主講（這是講課時的最新版本）。

市場主流消息中間件對比

Kafka中的基本概念

消息和批次

消息 ，Kafka里的數據單元，也就是我們一般消息中間件里的消息的概念（可以比作數據庫中一條記錄）。消息由字節(jié)數組組成。消息還可以包含鍵（可選元數據，也是字節(jié)數組），主要用于對消息選取分區(qū)。

作為一個高效的消息系統，為了提高效率，消息可以被分批寫入Kafka。批次就是一組消息，這些消息屬于同一個主題和分區(qū)。如果只傳遞單個消息，會導致大量的網絡開銷，把消息分成批次傳輸可以減少這開銷。但是，這個需要權衡（時間延遲和吞吐量之間），批次里包含的消息越多，單位時間內處理的消息就越多，單個消息的傳輸時間就越長（吞吐量高延時也高）。如果進行壓縮，可以提升數據的傳輸和存儲能力，但需要更多的計算處理。

對于Kafka來說，消息是晦澀難懂的字節(jié)數組，一般我們使用序列化和反序列化技術，格式常用的有JSON和XML，還有Avro（Hadoop開發(fā)的一款序列化框架），具體怎么使用依據自身的業(yè)務來定。

主題和分區(qū)

Kafka里的消息用主題進行分類（主題好比數據庫中的表），主題下有可以被分為若干個 分區(qū)（分表技術） 。分區(qū)本質上是個提交日志文件，有新消息，這個消息就會以追加的方式寫入分區(qū)（寫文件的形式），然后用先入先出的順序讀取。

[外鏈圖片轉存失敗,源站可能有防盜鏈機制,建議將圖片保存下來直接上傳(img-PG8I02O1-1686193157434)(file:///C:\Users\ADMINI~1\AppData\Local\Temp\msohtmlclip1\01\clip_image002.jpg)]

但是因為主題會有多個分區(qū)，所以在整個主題的范圍內，是無法保證消息的順序的，單個分區(qū)則可以保證。

Kafka通過分區(qū)來實現數據冗余和伸縮性，因為分區(qū)可以分布在不同的服務器上，那就是說一個主題可以跨越多個服務器（這是Kafka高性能的一個原因，多臺服務器的磁盤讀寫性能比單臺更高）。

前面我們說Kafka可以看成一個流平臺，很多時候，我們會把一個主題的數據看成一個流，不管有多少個分區(qū)。

生產者和消費者、偏移量、消費者群組

就是一般消息中間件里生產者和消費者的概念。一些其他的高級客戶端API，像數據管道API和流式處理的Kafka Stream，都是使用了最基本的生產者和消費者作為內部組件，然后提供了高級功能。

生產者默認情況下把消息均衡分布到主題的所有分區(qū)上，如果需要指定分區(qū)，則需要使用消息里的消息鍵和分區(qū)器。

消費者訂閱主題，一個或者多個，并且按照消息的生成順序讀取。消費者通過檢查所謂的偏移量來區(qū)分消息是否讀取過。偏移量是一種元數據，一個不斷遞增的整數值，創(chuàng)建消息的時候，Kafka會把他加入消息。在一個主題中一個分區(qū)里，每個消息的偏移量是唯一的。每個分區(qū)最后讀取的消息偏移量會保存到Zookeeper或者Kafka上，這樣分區(qū)的消費者關閉或者重啟，讀取狀態(tài)都不會丟失。

多個消費者可以構成一個消費者群組。怎么構成？共同讀取一個主題的消費者們，就形成了一個群組。群組可以保證每個分區(qū)只被一個消費者使用。

消費者和分區(qū)之間的這種映射關系叫做消費者對分區(qū)的所有權關系，很明顯，一個分區(qū)只有一個消費者，而一個消費者可以有多個分區(qū)。

（吃飯的故事：一桌一個分區(qū)，多桌多個分區(qū)，生產者不斷生產消息(消費)，消費者就是買單的人，消費者群組就是一群買單的人），一個分區(qū)只能被消費者群組中的一個消費者消費（不能重復消費），如果有一個消費者掛掉了<James跑路了>，另外的消費者接上）

Broker和集群

一個獨立的Kafka服務器叫Broker。broker的主要工作是，接收生產者的消息，設置偏移量，提交消息到磁盤保存；為消費者提供服務，響應請求，返回消息。在合適的硬件上，單個broker可以處理上千個分區(qū)和每秒百萬級的消息量。（要達到這個目的需要做操作系統調優(yōu)和JVM調優(yōu)）

多個broker可以組成一個集群。每個集群中broker會選舉出一個集群控制器?？刂破鲿M行管理，包括將分區(qū)分配給broker和監(jiān)控broker。

集群里，一個分區(qū)從屬于一個broker，這個broker被稱為首領。但是分區(qū)可以被分配給多個broker，這個時候會發(fā)生分區(qū)復制。

集群中Kafka內部一般使用管道技術進行高效的復制。

[外鏈圖片轉存失敗,源站可能有防盜鏈機制,建議將圖片保存下來直接上傳(img-trE4nSw7-1686193157437)(file:///C:\Users\ADMINI~1\AppData\Local\Temp\msohtmlclip1\01\clip_image006.jpg)]

分區(qū)復制帶來的好處是，提供了消息冗余。一旦首領broker失效，其他broker可以接管領導權。當然相關的消費者和生產者都要重新連接到新的首領上。

保留消息

在一定期限內保留消息是Kafka的一個重要特性，Kafka broker默認的保留策略是：要么保留一段時間（7天），要么保留一定大小（比如1個G）。到了限制，舊消息過期并刪除。但是每個主題可以根據業(yè)務需求配置自己的保留策略（開發(fā)時要注意，Kafka不像Mysql之類的永久存儲）。

為什么選擇Kafka

優(yōu)點

多生產者和多消費者

基于磁盤的數據存儲，換句話說，Kafka的數據天生就是持久化的。

高伸縮性，Kafka一開始就被設計成一個具有靈活伸縮性的系統，對在線集群的伸縮絲毫不影響整體系統的可用性。

高性能，結合橫向擴展生產者、消費者和broker，Kafka可以輕松處理巨大的信息流（LinkedIn公司每天處理萬億級數據），同時保證亞秒級的消息延遲。

常見場景

活動跟蹤

跟蹤網站用戶和前端應用發(fā)生的交互，比如頁面訪問次數和點擊，將這些信息作為消息發(fā)布到一個或者多個主題上，這樣就可以根據這些數據為機器學習提供數據，更新搜素結果等等（頭條、淘寶等總會推送你感興趣的內容，其實在數據分析之前就已經做了活動跟蹤）。

傳遞消息

標準消息中間件的功能

收集指標和日志

收集應用程序和系統的度量監(jiān)控指標，或者收集應用日志信息，通過Kafka路由到專門的日志搜索系統，比如ES。（國內用得較多）

提交日志

收集其他系統的變動日志，比如數據庫?？梢园褦祿斓母掳l(fā)布到Kafka上，應用通過監(jiān)控事件流來接收數據庫的實時更新，或者通過事件流將數據庫的更新復制到遠程系統。

還可以當其他系統發(fā)生了崩潰，通過重放日志來恢復系統的狀態(tài)。（異地災備）

流處理

操作實時數據流，進行統計、轉換、復雜計算等等。隨著大數據技術的不斷發(fā)展和成熟，無論是傳統企業(yè)還是互聯網公司都已經不再滿足于離線批處理，實時流處理的需求和重要性日益增長
。

近年來業(yè)界一直在探索實時流計算引擎和API，比如這幾年火爆的Spark
Streaming、Kafka Streaming、Beam和Flink，其中阿里雙11會場展示的實時銷售金額，就用的是流計算，是基于Flink，然后阿里在其上定制化的Blink。

Kafka的安裝、管理和配置

安裝

預備環(huán)境

Kafka是Java生態(tài)圈下的一員，用Scala編寫，運行在Java虛擬機上，所以安裝運行和普通的Java程序并沒有什么區(qū)別。

安裝Kafka官方說法，Java環(huán)境推薦Java8。

Kafka需要Zookeeper保存集群的元數據信息和消費者信息。Kafka一般會自帶Zookeeper，但是從穩(wěn)定性考慮，應該使用單獨的Zookeeper，而且構建Zookeeper集群。

運行

Kafka with ZooKeeper

啟動Zookeeper

進入Kafka目錄下的bin\windows

執(zhí)行kafka-server-start.bat …/…/config/server.properties

Linux下與此類似，進入bin后，執(zhí)行對應的sh文件即可

Kafka with KRaft

1、生成集群id

2、格式化存儲目錄

3、啟動服務

啟動正確后的界面如下：

kafka基本的操作和管理

## 列出所有主題

./kafka-topics.sh --bootstrap-server localhost:9092 --list

## 列出所有主題的詳細信息

./kafka-topics.sh --bootstrap-server localhost:9092 --describe

## 創(chuàng)建主題主題名 my-topic ，1副本，8分區(qū)

./kafka-topics.sh --bootstrap-server localhost:9092  --create --topic my-topic --replication-factor 1 --partitions 8

## 增加分區(qū)，注意：分區(qū)無法被刪除

./kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic my-topic --partitions 16

## 創(chuàng)建生產者（控制臺）

./kafka-console-producer.sh --broker-list localhost:9092 --topic my-topic

## 創(chuàng)建消費者（控制臺）

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --from-beginning --consumer.config ../config/consumer.properties

## kafka終止命令

./kafka-server-stop.sh 

總結就是：

Broker配置

配置文件放在Kafka目錄下的config目錄中，主要是server.properties文件

常規(guī)配置

broker.id

在單機時無需修改，但在集群下部署時往往需要修改。它是個每一個broker在集群中的唯一表示，要求是正數。當該服務器的IP地址發(fā)生改變時，broker.id沒有變化，則不會影響consumers的消息情況

listeners

監(jiān)聽列表(以逗號分隔 不同的協議(如plaintext,trace,ssl、不同的IP和端口)),hostname如果設置為0.0.0.0則綁定所有的網卡地址；如果hostname為空則綁定默認的網卡。如果沒有配置則默認為java.net.InetAddress.getCanonicalHostName()。

如：PLAINTEXT://myhost:9092,TRACE://:9091或 PLAINTEXT://0.0.0.0:9092,

zookeeper.connect

zookeeper集群的地址，可以是多個，多個之間用逗號分割。（一組hostname:port/path列表,hostname是zk的機器名或IP、port是zk的端口、/path是可選zk的路徑，如果不指定，默認使用根路徑）

log.dirs

Kafka把所有的消息都保存在磁盤上，存放這些數據的目錄通過log.dirs指定?？梢允褂枚嗦窂?#xff0c;使用逗號分隔。如果是多路徑，Kafka會根據“最少使用”原則，把同一個分區(qū)的日志片段保存到同一路徑下。會往擁有最少數據分區(qū)的路徑新增分區(qū)。

num.recovery.threads.per.data.dir

每數據目錄用于日志恢復啟動和關閉時的線程數量。因為這些線程只是服務器啟動（正常啟動和崩潰后重啟）和關閉時會用到。所以完全可以設置大量的線程來達到并行操作的目的。注意，這個參數指的是每個日志目錄的線程數，比如本參數設置為8，而log.dirs設置為了三個路徑，則總共會啟動24個線程。

auto.create.topics.enable

是否允許自動創(chuàng)建主題。如果設為true，那么produce（生產者往主題寫消息），consume（消費者從主題讀消息）或者fetch
metadata（任意客戶端向主題發(fā)送元數據請求時）一個不存在的主題時，就會自動創(chuàng)建。缺省為true。

delete.topic.enable=true

刪除主題配置，默認未開啟

主題配置

新建主題的默認參數

num.partitions

每個新建主題的分區(qū)個數（分區(qū)個數只能增加，不能減少 ）。這個參數一般要評估，比如，每秒鐘要寫入和讀取1000M數據，如果現在每個消費者每秒鐘可以處理50MB的數據，那么需要20個分區(qū)，這樣就可以讓20個消費者同時讀取這些分區(qū)，從而達到設計目標。（一般經驗，把分區(qū)大小限制在25G之內比較理想）

log.retention.hours

日志保存時間，默認為7天（168小時）。超過這個時間會清理數據。bytes和minutes無論哪個先達到都會觸發(fā)。與此類似還有l(wèi)og.retention.minutes和log.retention.ms，都設置的話，優(yōu)先使用具有最小值的那個。（提示：時間保留數據是通過檢查磁盤上日志片段文件的最后修改時間來實現的。也就是最后修改時間是指日志片段的關閉時間，也就是文件里最后一個消息的時間戳）

log.retention.bytes

topic每個分區(qū)的最大文件大小，一個topic的大小限制 = 分區(qū)數*log.retention.bytes。-1沒有大小限制。log.retention.bytes和log.retention.minutes任意一個達到要求，都會執(zhí)行刪除。(注意如果是log.retention.bytes先達到了，則是刪除多出來的部分數據)，一般不推薦使用最大文件刪除策略，而是推薦使用文件過期刪除策略。

log.segment.bytes

分區(qū)的日志存放在某個目錄下諸多文件中，這些文件將分區(qū)的日志切分成一段一段的，我們稱為日志片段。這個屬性就是每個文件的最大尺寸；當尺寸達到這個數值時，就會關閉當前文件，并創(chuàng)建新文件。被關閉的文件就開始等待過期。默認為1G。

如果一個主題每天只接受100MB的消息，那么根據默認設置，需要10天才能填滿一個文件。而且因為日志片段在關閉之前，消息是不會過期的，所以如果log.retention.hours保持默認值的話，那么這個日志片段需要17天才過期。因為關閉日志片段需要10天，等待過期又需要7天。

[外鏈圖片轉存失敗,源站可能有防盜鏈機制,建議將圖片保存下來直接上傳(img-15U5n3yQ-1686193157450)(file:///C:\Users\ADMINI~1\AppData\Local\Temp\msohtmlclip1\01\clip_image010.jpg)]

log.segment.ms

作用和log.segment.bytes類似，只不過判斷依據是時間。同樣的，兩個參數，以先到的為準。這個參數默認是不開啟的。

message.max.bytes

表示一個服務器能夠接收處理的消息的最大字節(jié)數，注意這個值producer和consumer必須設置一致，且不要大于fetch.message.max.bytes屬性的值(消費者能讀取的最大消息,這個值應該大于或等于message.max.bytes)。該值默認是1000000字節(jié)，大概900KB~1MB。如果啟動壓縮，判斷壓縮后的值。這個值的大小對性能影響很大，值越大，網絡和IO的時間越長，還會增加磁盤寫入的大小。

Kafka設計的初衷是迅速處理短小的消息，一般10K大小的消息吞吐性能最好（LinkedIn的kafka性能測試）

硬件配置對Kafka性能的影響

為Kafka選擇合適的硬件更像是一門藝術，就跟它的名字一樣，我們分別從磁盤、內存、網絡和CPU上來分析，確定了這些關注點，就可以在預算范圍之內選擇最優(yōu)的硬件配置。

磁盤吞吐量/磁盤容量

磁盤吞吐量（IOPS 每秒的讀寫次數）會影響生產者的性能。因為生產者的消息必須被提交到服務器保存，大多數的客戶端都會一直等待，直到至少有一個服務器確認消息已經成功提交為止。也就是說，磁盤寫入速度越快，生成消息的延遲就越低。（SSD固態(tài)貴單個速度快，HDD機械偏移可以多買幾個，設置多個目錄加快速度，具體情況具體分析）

磁盤容量的大小，則主要看需要保存的消息數量。如果每天收到1TB的數據，并保留7天，那么磁盤就需要7TB的數據。

內存

Kafka本身并不需要太大內存，內存則主要是影響消費者性能。在大多數業(yè)務情況下，消費者消費的數據一般會從內存（頁面緩存，從系統內存中分）中獲取，這比在磁盤上讀取肯定要快的多。一般來說運行Kafka的JVM不需要太多的內存，剩余的系統內存可以作為頁面緩存，或者用來緩存正在使用的日志片段，所以我們一般Kafka不會同其他的重要應用系統部署在一臺服務器上，因為他們需要共享頁面緩存，這個會降低Kafka消費者的性能。

[外鏈圖片轉存失敗,源站可能有防盜鏈機制,建議將圖片保存下來直接上傳(img-vsyB4Z1r-1686193157452)(file:///C:\Users\ADMINI~1\AppData\Local\Temp\msohtmlclip1\01\clip_image012.jpg)]

網絡

網絡吞吐量決定了Kafka能夠處理的最大數據流量。它和磁盤是制約Kafka拓展規(guī)模的主要因素。對于生產者、消費者寫入數據和讀取數據都要瓜分網絡流量。同時做集群復制也非常消耗網絡。

CPU

Kafka對cpu的要求不高，主要是用在對消息解壓和壓縮上。所以cpu的性能不是在使用Kafka的首要考慮因素。

總結

我們要為Kafka選擇合適的硬件時，優(yōu)先考慮存儲，包括存儲的大小，然后考慮生產者的性能（也就是磁盤的吞吐量），選好存儲以后，再來選擇CPU和內存就容易得多。網絡的選擇要根據業(yè)務上的情況來定，也是非常重要的一環(huán)。