一、概要

Kafka作為一種高吞吐量的分布式發(fā)布訂閱消息系統(tǒng)，在消息應(yīng)用中廣泛使用，尤其在需要實(shí)時(shí)數(shù)據(jù)處理和應(yīng)用程序活動(dòng)跟蹤的場(chǎng)景，kafka已成為首選服務(wù)；在Kafka2.8之前，Kafka強(qiáng)依賴zookeeper來(lái)來(lái)負(fù)責(zé)集群元數(shù)據(jù)的管理，這也導(dǎo)致當(dāng)Zookeeper集群性能發(fā)生抖動(dòng)時(shí)，Kafka的性能也會(huì)收到很大的影響。2.8版本之后，kafka3.x開始提供KRaft（Kafka Raft，依賴Java 8+ ）模式，開始去除對(duì)zookeeper的依賴。最新的3.5版本中，Kafka依然兼容zookeeper Controller，但Kafka Raft元數(shù)據(jù)模式，已經(jīng)可以在不依賴zookeeper的情況下獨(dú)立啟動(dòng)Kafka了。

kraft模式的優(yōu)勢(shì)：

1、更簡(jiǎn)單的部署和管理——通過(guò)只安裝和管理一個(gè)應(yīng)用程序，Kafka 現(xiàn)在的運(yùn)營(yíng)足跡要小得多。這也使得在邊緣的小型設(shè)備中更容易利用 Kafka；
2、提高可擴(kuò)展性——KRaft 的恢復(fù)時(shí)間比 ZooKeeper 快一個(gè)數(shù)量級(jí)。這使我們能夠有效地?cái)U(kuò)展到單個(gè)集群中的數(shù)百萬(wàn)個(gè)分區(qū)。ZooKeeper 的有效限制是數(shù)萬(wàn)；
3、更有效的元數(shù)據(jù)傳播——基于日志、事件驅(qū)動(dòng)的元數(shù)據(jù)傳播可以提高 Kafka 的許多核心功能的性能。另外它還支持元數(shù)據(jù)主題的快照。

資料鏈接：官網(wǎng)kraft、kafka_jira

二、拓?fù)?/h2>

2.1、早期拓?fù)?/h3>

老controller選舉原理： 只有一個(gè)broker可以當(dāng)選Controller，當(dāng)Controller在的broker宕機(jī)后，其他Broker可以競(jìng)爭(zhēng)當(dāng)選Controller。

Kafka是通過(guò)使用epoch number（紀(jì)元編號(hào)，也稱為隔離令牌）來(lái)完成的。epoch number只是單調(diào)遞增的數(shù)字，第一次選出Controller時(shí)，epoch number值為1，如果再次選出新的Controller，則epoch number將為2，依次單調(diào)遞增。

每個(gè)新選出的controller通過(guò)Zookeeper 的條件遞增操作獲得一個(gè)全新的、數(shù)值更大的epoch number 。其他Broker 在知道當(dāng)前epoch number 后，如果收到由controller發(fā)出的包含較舊(較小)epoch number的消息，就會(huì)忽略它們，即Broker根據(jù)最大的epoch number來(lái)區(qū)分當(dāng)前最新的controller。

常見幾種副本狀態(tài)：

New：controller剛剛創(chuàng)建副本時(shí)副本的狀態(tài)，此狀態(tài)的副本只能成為follower副本
Online：啟動(dòng)副本后變更為該狀態(tài)，此狀態(tài)的副本既可以成為follower副本也可以成為leader副本
Offline：當(dāng)副本所在的broker宕機(jī)后，副本的狀態(tài)就會(huì)轉(zhuǎn)變?yōu)镺ffline
ReplicaDeletionStarted：當(dāng)Kafa集群開啟了topic刪除且收到某個(gè)topic的刪除命令時(shí)，該topic下的副本就會(huì)進(jìn)入該狀態(tài)
ReplicaDeletionSuccessful：當(dāng)副本成功刪除以后，副本就會(huì)進(jìn)入該狀態(tài)
ReplicaDeletionIneligible：當(dāng)有副本刪除失敗時(shí)，副本就會(huì)進(jìn)入該狀態(tài)，等待controller的重試
NonExistent：當(dāng)副本成功刪除后，副本進(jìn)入該狀態(tài)，或者Topic剛剛創(chuàng)建副本并未建立時(shí)，副本就處于該狀態(tài)。

當(dāng)Kafka Topic剛剛創(chuàng)建時(shí)，該Topic的副本處于NonExistent狀態(tài)，此時(shí)controller加載Zookeeper中該Topic每個(gè)分區(qū)的副本信息到內(nèi)存中，同時(shí)將副本更新為New狀態(tài)，之后controller選擇分區(qū)的第一個(gè)副本作為leader副本并設(shè)置所有副本進(jìn)入ISR，然后再Kafka中持久化該決定。

當(dāng)確定了分區(qū)副本以及l(fā)eader以后，controller會(huì)將這些信息發(fā)送給各個(gè)副本，同時(shí)將副本狀態(tài)同步給所有的broker。上述操作完成以后，副本就會(huì)進(jìn)入Online狀態(tài)。

當(dāng)開啟了Topic刪除操作，controller會(huì)停止所有副本，如果是follower副本將停止向leader副本fetch數(shù)據(jù)，如果是leader副本，controller會(huì)設(shè)置該分為的leade為NO_LEADER，之后副本進(jìn)入Offline狀態(tài)。緊接著，controller會(huì)將副本的狀態(tài)變?yōu)镽eplicaDeletionStarted狀態(tài)表明開始進(jìn)行Topic刪除。controller會(huì)向所有的副本發(fā)出刪除請(qǐng)求，請(qǐng)求他們刪除本地的副本數(shù)據(jù)，當(dāng)所有副本刪除成功以后，便會(huì)進(jìn)入ReplicaDeletionSuccessful狀態(tài)。假設(shè)刪除過(guò)程中有其中一個(gè)副本刪除失敗，則會(huì)進(jìn)入ReplicaDeletionIneligible狀態(tài)，等待controller的重試。同時(shí)處于ReplicaDeletionSuccessful狀態(tài)會(huì)自動(dòng)變?yōu)镹onExistent狀態(tài)，同時(shí)controller的上下文緩存會(huì)清除這些副本信息。

以下是分區(qū)狀態(tài)：

NonExistent：表明分區(qū)不存在或分區(qū)已被刪除
New：一旦分區(qū)被創(chuàng)建，分區(qū)便處于該狀態(tài)。此時(shí)Kafka已經(jīng)確定了分區(qū)列表，但是還沒(méi)有選出Leader和ISR。
Online：一旦分區(qū)leader被選出，則進(jìn)入該狀態(tài)，表示分區(qū)可以正常工作
Offline：當(dāng)分區(qū)leader所在的broker宕機(jī)以后則會(huì)進(jìn)入該狀態(tài)，表明分區(qū)無(wú)法正常工作

當(dāng)創(chuàng)建Topic時(shí)，controller負(fù)責(zé)創(chuàng)建分區(qū)對(duì)象，首先會(huì)短暫的將所有分區(qū)置為NonExistent狀態(tài)，之后讀取Zookeeper中的副本分配方案，然后令分區(qū)進(jìn)入New狀態(tài)。處于New狀態(tài)的分區(qū)當(dāng)選出Leader副本和ISR時(shí)，將會(huì)進(jìn)入Online狀態(tài)。

若用戶發(fā)起了刪除Topic操作，分區(qū)會(huì)進(jìn)入NonExistent狀態(tài)，并且還會(huì)開啟分區(qū)下的副本刪除操作。如果是關(guān)閉broker操作或者宕機(jī)，controller會(huì)判斷broker是否是分區(qū)leader，如果是則需要開啟新一輪的分區(qū)leader選舉然后將分區(qū)狀態(tài)改回Online狀態(tài)。

2.2、kRaft controller拓?fù)?/h3>

kafka kraft模式集群：必須是奇數(shù)節(jié)點(diǎn)，3節(jié)點(diǎn)最大容錯(cuò)1個(gè)，5節(jié)點(diǎn)容錯(cuò)2個(gè)

KRaft集群中，所有控制器代理都維護(hù)一個(gè)保持最新的內(nèi)存元數(shù)據(jù)緩存，以便任何控制器都可以在需要時(shí)接管作為活動(dòng)控制器（active controller），所有brokers都與控制器進(jìn)行通信，其中活動(dòng)控制器將處理與其他brokers通信對(duì)元數(shù)據(jù)的更改。KRaft 基于 Raft 共識(shí)協(xié)議，該協(xié)議作為 KIP-500 的一部分引入 Kafka，并在其他相關(guān) KIP 中定義了更多細(xì)節(jié)?；顒?dòng)控制器是kafka kraft集群內(nèi)部元數(shù)據(jù)主題的單個(gè)分區(qū)的leader，其他控制器是副本follower，brokers是副本觀察者。因此，不是controlle將元數(shù)據(jù)更改廣播給其他控制器或brokers，而是它們各自主動(dòng)獲取更改。這使得保持所有控制器和brokers同步非常有效，并且還縮短了broker和控制器的重啟時(shí)間。

在 KRaft 模式下，集群元數(shù)據(jù)（反映所有控制器管理資源的當(dāng)前狀態(tài)）存儲(chǔ)在名為__cluster_metadata. KRaft 使用這個(gè)主題在控制器和代理節(jié)點(diǎn)之間同步集群狀態(tài)更改。KRaft 模式下，Kafka 集群可以以專用或共享模式運(yùn)行。在專用模式下，一些節(jié)點(diǎn)將其process.roles配置設(shè)置為controller，而其余節(jié)點(diǎn)將其設(shè)置為broker。對(duì)于共享模式，這些節(jié)點(diǎn)將process.roles設(shè)置為controller, broker，即些節(jié)點(diǎn)將執(zhí)行雙重任務(wù)。該模式中那個(gè)稱為“controller”的特殊節(jié)點(diǎn)負(fù)責(zé)管理集群中代理的注冊(cè)，它一般就是controller.quorum.voters中的第一個(gè)節(jié)點(diǎn)。Broker 活躍度有兩個(gè)條件：

1、Brokers must maintain an active session with the controller in order to receive regular metadata updates.而 “active session” depends on the cluster configuration，an active session is maintained by sending periodic heartbeats to the controller. If the controller fails to receive a heartbeat before the timeout configured by broker.session.timeout.ms expires, then the node is considered offline.
2、Brokers acting as followers must replicate the writes from the leader and not fall “too far” behind.

Kradt模式下，the controllers store the cluster metadata in the directory specified in metadata.log.dir/the first log directory。特別注意，新增controller節(jié)點(diǎn)時(shí)需等待現(xiàn)有的controller提交完成所有數(shù)據(jù)后：The new controller node should not be formatted and started until the majority of the controllers have all of the committed data. 可通過(guò)如下命令來(lái)檢查確認(rèn)：kafka-metadata-quorum.sh --bootstrap-server broker_host:port describe --replication，如果顯示Lag的值為0就最好，至少保證Lag值對(duì)于controllers足夠小，或如果leader’s end offset is not increasing, you can wait until the lag is 0 for a majority；如果不是0，要檢查下LastFetchTimestamp 和 LastCaughtUpTimestamp這2個(gè)值，所有controller的這2個(gè)值要盡量接近，即讀到的和消費(fèi)掉的要一致；之后才能執(zhí)行：bin/kafka-storage.sh format --cluster-id uuid --config server_properties對(duì)新controller的元數(shù)據(jù)存儲(chǔ)路徑進(jìn)行格式化；注意，混合模式下，存儲(chǔ)格式化會(huì)報(bào)錯(cuò)：Log directory … is already formatted，這中情況僅出現(xiàn)在混合模式下且原controller 日志路徑丟失或損壞的情況下，可執(zhí)行：bin/kafka-storage.sh format --cluster-id uuid --ignore-formatted --config server_properties ，其他情況不建議使用?？刂破饔脕?lái)接收來(lái)自其他控制器和borkers的請(qǐng)求。因此，即使服務(wù)器沒(méi)有啟用controller角色（即它只是一個(gè)broker），它仍然必須定義控制器偵聽器以及配置它所需的任何安全屬性。配置參考示例：

process.roles=broker
listeners=BROKER://localhost:9092  #broker does not expose the controller listener itself
inter.broker.listener.name=BROKER  #區(qū)別于下面的controller.listener.names
controller.quorum.voters=0@localhost:9093
controller.listener.names=CONTROLLER  #僅用于controller
listener.security.protocol.map=BROKER:SASL_SSL,CONTROLLER:SASL_SSL#混合模式
process.roles=broker,controller
listeners=BROKER://localhost:9092,CONTROLLER://localhost:9093  #listeners獨(dú)立配置用戶kafka 客戶端訪問(wèn)
inter.broker.listener.name=BROKER  #配合上面的與kafka client交互，隔離開controller
controller.quorum.voters=0@localhost:9093
controller.listener.names=CONTROLLER   #The controller will accept requests on all listeners defined by controller.listener.names，多個(gè)controller時(shí)，the first one in the list will be used for outbound requests.
listener.security.protocol.map=BROKER:SASL_SSL,CONTROLLER:SASL_SSL

更多參見 KRaft Principal Forwarding。注意：kraft模式下的混合模式多用于開發(fā)環(huán)境，生產(chǎn)環(huán)境不建議使用?；旌夏Ｊ较乱粋€(gè)明顯的問(wèn)題就是， controller will be less isolated from the rest of the system，即無(wú)法與broker隔離，一個(gè)典型場(chǎng)景就是無(wú)法單獨(dú)對(duì)controller進(jìn)行彈性擴(kuò)展或回滾升級(jí)；KRaft模式下，特定的Kafka服務(wù)器將被選作為controller；所有被定義候選的controller都會(huì)參與元數(shù)據(jù)的競(jìng)選，對(duì)已a(bǔ)ctive的controller，其他的controller都充當(dāng)hot standby 角色；因此，我們常用做法配置process.role 為 broker/controller而不是兩者都具備（混合模式）；推薦，一個(gè)Kafka集群應(yīng)該使用3個(gè)controller就可，超過(guò)3個(gè)也不推薦。Kafka控制器將集群的所有元數(shù)據(jù)存儲(chǔ)在內(nèi)存和磁盤上，官方建議知識(shí)給內(nèi)存和磁盤分出各5G空間用來(lái)存儲(chǔ)這些metadata日志；另外，kraft也有一些限制問(wèn)題，比如以下功能未在KRaft模式中完全實(shí)現(xiàn)：

1、通過(guò)管理API配置SCRAM用戶
2、支持具有多個(gè)存儲(chǔ)目錄的JBOD配置
3、修改獨(dú)立KRaft控制器上的某些動(dòng)態(tài)配置
4、委派令牌tokens

KRaft元數(shù)據(jù)復(fù)制過(guò)程：

集群元數(shù)據(jù)存儲(chǔ)在 Kafka 主題中，而活動(dòng)控制器是元數(shù)據(jù)主題的單個(gè)分區(qū)的領(lǐng)導(dǎo)者，它將接收所有數(shù)據(jù)并寫入。其他控制器作為follower，將主動(dòng)獲取這些更改。對(duì)比傳統(tǒng)的副本復(fù)制，當(dāng)需要選舉新leader時(shí)，是通過(guò)仲裁完成的，而不是同步副本集。因此，元數(shù)據(jù)復(fù)制不涉及 ISR。另一個(gè)區(qū)別是元數(shù)據(jù)記錄在寫入每個(gè)節(jié)點(diǎn)的本地日志時(shí)會(huì)立即刷新到磁盤。

Kraft 模式 以及 安全認(rèn)證配置：

SASL/GSSAPI：kerberos認(rèn)證方式，一般使用隨機(jī)密碼的keytab認(rèn)證方式，密碼是加密的，也是企業(yè)里使用最多的認(rèn)證方式；
SASL/PLAIN：這種方式其實(shí)就是一個(gè)賬號(hào)/密碼的認(rèn)證方式，不過(guò)它有很多缺陷，比如用戶名密碼是存儲(chǔ)在文件中，不能動(dòng)態(tài)添加，密碼明文等等！好處是足夠簡(jiǎn)單；
SASL/SCRAM：針對(duì)SASL/PLAIN方式的不足而提供的另一種認(rèn)證方式。這種方式的用戶名/密碼是存儲(chǔ)中zookeeper的，因此能夠支持動(dòng)態(tài)添加用戶。該種認(rèn)證方式還會(huì)使用sha256或sha512對(duì)密碼加密，安全性相對(duì)會(huì)高一些，在0.10.2版本引入；

三、部署配置

wget https://archive.apache.org/dist/kafka/3.4.0/kafka_2.13-3.4.0.tgz
tar -zxvf kafka_2.13-3.4.0.tgz -C /opt/
mv /opt/kafka_2.13-3.4.0. /opt/kafka
chown kafka:kafka -R /opt/kafka
cd /opt/kafka/
mkdir data
vim /opt/kafka/config/kraftserver.properties  //如下所示# The role of this server. Setting this puts us in KRaft mode
process.roles=broker,controller   #the server acts as both a broker and a controller# The node id associated with this instance's roles
node.id=2# The connect string for the controller quorum 集群選舉控制器配置，默認(rèn)走 PLAINTEXT協(xié)議除非顯示定義其他協(xié)議
controller.quorum.voters=1@172.18.1.176:9093,2@172.18.1.217:9093,3@172.18.1.150:9093############################# Socket Server Settings ############################## The address the socket server listens on.
# Combined nodes (i.e. those with `process.roles=broker,controller`) must list the controller listener here at a minimum.
# If the broker listener is not defined, the default listener will use a host name that is equal to the value of java.net.InetAddress.getCanonicalHostName(),
# with PLAINTEXT listener name, and port 9092.
#   FORMAT:
#     listeners = listener_name://host_name:port
#   EXAMPLE:
#     listeners = PLAINTEXT://your.host.name:9092
listeners=PLAINTEXT://172.31.7.237:9092,CONTROLLER://172.18.1.217:9093# Name of listener used for communication between brokers. it is used exclusively僅用于 for requests between brokers
inter.broker.listener.name=PLAINTEXT   #注意與controller.listener.names不要沖突# Listener name, hostname and port the broker will advertise to clients.
# If not set, it uses the value for "listeners".
#advertised.listeners=PLAINTEXT://172.18.1.217:9092#完成后，生成整個(gè)集群有一個(gè)唯一的ID標(biāo)志，使用uuid?？墒褂霉俜教峁┑?kafka-storage 工具生成
/opt/kafka/bin/kafka-storage.sh random-uuid
或
KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"#用上述ID格式化存儲(chǔ)路徑
/opt/kafka/bin/kafka-storage.sh format -t clust_ID -c /opt/kafka/config/kraft/server.properties
或
bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c config/kraft/server.properties
#完成后以kraft模式啟動(dòng)服務(wù)
bin/kafka-server-start.sh -daemon ./config/kraft/server.properties#創(chuàng)建topic
bin/kafka-topics.sh --create --topic First_Kafka_Topic --partitions 1 --replication-factor 3 --bootstrap-server 172.31.7.237:9092#查看
bin/kafka-topics.sh --list --bootstrap-server 172.31.7.237:9092

2）kafka啟動(dòng)腳本

#!/bin/bash
#kafka集群?jiǎn)?dòng)腳本
case $1 in"start"){for i in 172.18.1.176,172.18.1.217,@172.18.1.150doecho "--------啟動(dòng) $i kafka with kraft-------"ssh $i "/home/kafka/bin/kafka-server-start.sh -daemon /home/kafka/config/kraft/server.properties"done
};;
"stop"){for i in 172.18.1.176,172.18.1.217,@172.18.1.150doecho "------停止 $i kafka--------"ssh $i "/home/kafka/bin/kafka-server-stop.sh"done
};;
esac

3）kafka kRaft SASL/PLAIN 安全認(rèn)證配置 ：

新建config 文件 config/kafka_server_jaas.conf，如下所示：

KafkaServer {
org.apache.kafka.common.security.plain.PlainLoginModule required
serviceName="kafka"
username="admin"
password="admin"
user_admin="admin";
};

復(fù)制一份kafka-server-start.sh ，修改名稱 kafka-server-start-sasl.sh啟動(dòng)腳本修改名稱，引入加密文件：

……
if [ "x$KAFKA_HEAP_OPTS" = "x" ]; thenexport KAFKA_HEAP_OPTS="-Xmx1G -Xms1G -Djava.security.auth.login.config=/opt/kafka/kafka_2.13-3.4.0/config/kafka_server_jaas.conf"
fi
……

復(fù)制一份 server_sasl.properties 修改名稱為server_sasl.properties 與非安全認(rèn)證分開，編輯：

node.id=1
controller.quorum.voters=1@kraft1:9093
listeners=SASL_PLAINTEXT://172.31.7.237:9092,CONTROLLER:///172.18.1.217:9093
inter.broker.listener.name=SASL_PLAINTEXT
advertised.listeners=SASL_PLAINTEXT://172.31.7.237:9092
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN

完成后啟動(dòng)kafka：sh ./bin/kafka-server-start-sasl.sh -daemon ./config/kraft/server_sasl.properties

四、附錄：知識(shí)回顧

4.1、常用Message Queue對(duì)比

1）RabbitMQ

RabbitMQ是使用Erlang編寫的一個(gè)開源的消息隊(duì)列，本身支持很多的協(xié)議：AMQP，XMPP, SMTP, STOMP，也正因如此，它非常重量級(jí)，更適合于企業(yè)級(jí)的開發(fā)。同時(shí)實(shí)現(xiàn)了Broker構(gòu)架，這意味著消息在發(fā)送給客戶端時(shí)先在中心隊(duì)列排隊(duì)。對(duì)路由，負(fù)載均衡或者數(shù)據(jù)持久化都有很好的支持。

2）Redis

Redis是一個(gè)基于Key-Value對(duì)的NoSQL數(shù)據(jù)庫(kù)，開發(fā)維護(hù)很活躍。雖然它是一個(gè)Key-Value數(shù)據(jù)庫(kù)存儲(chǔ)系統(tǒng)，但它本身支持MQ功能，所以完全可以當(dāng)做一個(gè)輕量級(jí)的隊(duì)列服務(wù)來(lái)使用。對(duì)于RabbitMQ和Redis的入隊(duì)和出隊(duì)操作，各執(zhí)行100萬(wàn)次，每10萬(wàn)次記錄一次執(zhí)行時(shí)間。測(cè)試數(shù)據(jù)分為128Bytes、512Bytes、1K和10K四個(gè)不同大小的數(shù)據(jù)。實(shí)驗(yàn)表明：入隊(duì)時(shí)，當(dāng)數(shù)據(jù)比較小時(shí)Redis的性能要高于RabbitMQ，而如果數(shù)據(jù)大小超過(guò)了10K，Redis則慢的無(wú)法忍受；出隊(duì)時(shí)，無(wú)論數(shù)據(jù)大小，Redis都表現(xiàn)出非常好的性能，而RabbitMQ的出隊(duì)性能則遠(yuǎn)低于Redis。

3）ZeroMQ

ZeroMQ號(hào)稱最快的消息隊(duì)列系統(tǒng)，尤其針對(duì)大吞吐量的需求場(chǎng)景。ZeroMQ能夠?qū)崿F(xiàn)RabbitMQ不擅長(zhǎng)的高級(jí)/復(fù)雜的隊(duì)列，但是開發(fā)人員需要自己組合多種技術(shù)框架，技術(shù)上的復(fù)雜度是對(duì)這MQ能夠應(yīng)用成功的挑戰(zhàn)。ZeroMQ具有一個(gè)獨(dú)特的非中間件的模式，你不需要安裝和運(yùn)行一個(gè)消息服務(wù)器或中間件，因?yàn)槟愕膽?yīng)用程序?qū)缪葸@個(gè)服務(wù)器角色。你只需要簡(jiǎn)單的引用ZeroMQ程序庫(kù)，可以使用NuGet安裝，然后你就可以愉快的在應(yīng)用程序之間發(fā)送消息了。但是ZeroMQ僅提供非持久性的隊(duì)列，也就是說(shuō)如果宕機(jī)，數(shù)據(jù)將會(huì)丟失。其中，Twitter的Storm 0.9.0以前的版本中默認(rèn)使用ZeroMQ作為數(shù)據(jù)流的傳輸（Storm從0.9版本開始同時(shí)支持ZeroMQ和Netty作為傳輸模塊）。

4）ActiveMQ

ActiveMQ是Apache下的一個(gè)子項(xiàng)目。 類似于ZeroMQ，它能夠以代理人和點(diǎn)對(duì)點(diǎn)的技術(shù)實(shí)現(xiàn)隊(duì)列。同時(shí)類似于RabbitMQ，它少量代碼就可以高效地實(shí)現(xiàn)高級(jí)應(yīng)用場(chǎng)景。

5）Kafka/Jafka

Kafka是Apache下的一個(gè)子項(xiàng)目，是一個(gè)高性能跨語(yǔ)言分布式發(fā)布/訂閱消息隊(duì)列系統(tǒng)，而Jafka是在Kafka之上孵化而來(lái)的，即Kafka的一個(gè)升級(jí)版。具有以下特性：快速持久化，可以在O(1)的系統(tǒng)開銷下進(jìn)行消息持久化；高吞吐，在一臺(tái)普通的服務(wù)器上既可以達(dá)到10W/s的吞吐速率；完全的分布式系統(tǒng)，Broker、Producer、Consumer都原生自動(dòng)支持分布式，自動(dòng)實(shí)現(xiàn)負(fù)載均衡；支持Hadoop數(shù)據(jù)并行加載，對(duì)于像Hadoop的一樣的日志數(shù)據(jù)和離線分析系統(tǒng)，但又要求實(shí)時(shí)處理的限制，這是一個(gè)可行的解決方案。Kafka通過(guò)Hadoop的并行加載機(jī)制統(tǒng)一了在線和離線的消息處理。Apache Kafka相對(duì)于ActiveMQ是一個(gè)非常輕量級(jí)的消息系統(tǒng)，除了性能非常好之外，還是一個(gè)工作良好的分布式系統(tǒng)。kafka3.0輕量級(jí)的單進(jìn)程部署不僅可以替代ActiveMQ、RabbitMQ等傳統(tǒng)的消息隊(duì)列，同時(shí)也適合于邊緣場(chǎng)景和使用輕量級(jí)硬件的場(chǎng)景。數(shù)據(jù)表明，一個(gè)可以管理 兩百萬(wàn)分區(qū)的集群中，kafka3.0新版本中Quorum Controller的遷移過(guò)程可以 從幾分鐘縮小至30秒，突破了限制Kafka集群范圍的元數(shù)據(jù)管理這個(gè)主要瓶頸，關(guān)機(jī)和重啟 十多倍速于kafka2.8，性能完全實(shí)現(xiàn)了強(qiáng)力碾壓！

4.2、Kafka的原理

Kafka的一種新的復(fù)制協(xié)議，它是在Kafka2.4版本中引入的。Kraft協(xié)議的目標(biāo)是提高Kafka的可靠性和可維護(hù)性，滿足分布式流處理需求；它主要包括以下幾個(gè)方面：

1、去中心化：Kraft 協(xié)議將Kafka 的控制器節(jié)點(diǎn)去中心化，每個(gè)副本都可以成為控制器，從而提高了系統(tǒng)的可靠性和容錯(cuò)性。

2、原子性：Kraft 協(xié)議使用原子廣播協(xié)議來(lái)保證消息的原子性，即所有副本都會(huì)收到相同的消息，從而保證了數(shù)據(jù)的一致性

3、可擴(kuò)展性：Kraft 協(xié)議支持動(dòng)態(tài)添加和刪除副本，從而提高了系統(tǒng)的可擴(kuò)展性和可維護(hù)性。

4、可靠性:Kraft 協(xié)議使用 Raft 算法來(lái)保證副本之間的一致性，從而提高了系統(tǒng)的可靠性和容錯(cuò)性。

4.3、DoctorKafka 管理工具

它是源于Pinterest產(chǎn)品的一個(gè)項(xiàng)目，為了擴(kuò)展Kafka 服務(wù)的運(yùn)維規(guī)模，Pinterest 構(gòu)建了DoctorKafka，用于管理 Kafka 集群自愈和工作負(fù)載均衡；。DoctorKafka 能夠探測(cè)到Kafka broker 的故障并自動(dòng)將故障broker 的負(fù)載轉(zhuǎn)移給健康的broker?，F(xiàn)在，Pinterest 已經(jīng)在 GitHub 上將該項(xiàng)目開源。DoctorKafka 由三部分組成,架構(gòu)如下：

1、部署在每個(gè) broker 上的指標(biāo)收集器（metrics collector），它會(huì)定期收集 Kafka 進(jìn)程和主機(jī)的指標(biāo)，并將其發(fā)布到一個(gè) Kafka 主題上。在這里，使用了 Kafka 作為 broker 的狀態(tài)存儲(chǔ)，這樣的話，能夠簡(jiǎn)化 DoctorKafka 的搭建過(guò)程并減少對(duì)其他系統(tǒng)的依賴；每個(gè) broker 上都會(huì)運(yùn)行一個(gè)指標(biāo)收集器，它會(huì)收集 Kafka broker 輸入和輸出的網(wǎng)絡(luò)流量指標(biāo)以及每個(gè)副本（replica）的狀態(tài)。
2、中心化的 DoctorKafka 服務(wù)會(huì)管理多個(gè)集群，分析 broker 的狀態(tài)指標(biāo)以探測(cè) broker 的故障，執(zhí)行集群自愈和負(fù)載均衡的命令。DoctorKafka 會(huì)將執(zhí)行的命令記錄在另外一個(gè)名為“Action Log”主題上；
3、用于瀏覽 Kafka 集群狀態(tài)和執(zhí)行流程的 Web UI 頁(yè)面。

DoctorKafka 服務(wù)啟動(dòng)之后，它會(huì)首先讀取 broker 最近 24 到 48 小時(shí)的狀態(tài)，基于此，DoctorKafka 會(huì)推斷每個(gè)副本工作負(fù)載所需的資源。因?yàn)?Kafka 工作負(fù)載主要是網(wǎng)絡(luò)密集型的，DoctorKafka 主要關(guān)注副本的網(wǎng)絡(luò)帶寬使用情況。DoctorKafka 在啟動(dòng)之后，會(huì)階段性地檢查每個(gè)集群的狀態(tài)。當(dāng)探測(cè)到 broker 出現(xiàn)故障時(shí)，它會(huì)將故障 broker 的工作負(fù)載轉(zhuǎn)移給有足夠帶寬的 broker。如果在集群中沒(méi)有足夠的資源進(jìn)行重分配的話，它會(huì)發(fā)出告警。與之類似，當(dāng) DoctorKafka 進(jìn)行工作負(fù)載平衡時(shí)，它會(huì)識(shí)別出網(wǎng)絡(luò)流量超出配置的 broker，并將工作負(fù)載轉(zhuǎn)移給流量更少的 broker，或者是執(zhí)行更優(yōu)的領(lǐng)導(dǎo)者選舉（leader election）方案來(lái)轉(zhuǎn)移流量。

DoctorKafka 已經(jīng)在 Pinterest 運(yùn)行了數(shù)月之久，并幫助其運(yùn)維人員管理著 1000 個(gè)以上的集群。更多參見 項(xiàng)目地址。

4.4、Kafka WebUI之Kowl

Kowl ( Kafka Owl)： Kafka WebUI for exploring messages, consumers. configurations and more with a focus on a good UI。更多參看：項(xiàng)目地址。

除上之外，還有9種常見的kafka UI工具，除此之外還有LogiKM 和 kafka-console-ui ，因Kafka-UI 功能齊全且免費(fèi)，推薦。

1 AKHQ 免費(fèi)
2 Kowl 部分收費(fèi)
3 Kafdrop 免費(fèi)
4 UI for Apache Kafka 免費(fèi)
5 Lenses 免費(fèi)
6 CMAK 免費(fèi)
7 Confluent CC 收費(fèi)
8 Conduktor 收費(fèi)
9 LogiKM 免費(fèi)
10 kafka-console-ui 免費(fèi)

一般該程序通過(guò)docker容器運(yùn)行，Kowl 的最大優(yōu)勢(shì)在于其出色的用戶界面。它方便、用戶友好且易于使用；啟動(dòng)后界面如下：

Kowl 提供消息瀏覽、實(shí)時(shí)跟蹤以及對(duì) Protobuf、Avro 和 Amazon MSK IAM 的支持,但登錄系統(tǒng)(Google、GitHub、Okta)和具有組同步的 RBAC 權(quán)限僅適用于付費(fèi)的 Kowl Business 計(jì)劃。Kowl 還缺少多集群管理、動(dòng)態(tài)主題配置、分區(qū)增加、副本更改、Kafka Connect 管理、模式注冊(cè)表、KSQL 集成、Kafka Streams 拓?fù)?、只讀模式以及 JMX 指標(biāo)的可視化和圖表等功能。