?1.Zookeeper集群的相關(guān)知識

?1.1 zookeeper的概念

? ? ? ?ZooKeeper是一個分布式的，開放源碼的分布式應(yīng)用程序協(xié)調(diào)服務(wù)，是Google的Chubby一個開源的實(shí)現(xiàn)，是Hadoop和Hbase的重要組件。它是一個為分布式應(yīng)用提供一致性服務(wù)的軟件，提供的功能包括：配置維護(hù)、域名服務(wù)、分布式同步、組服務(wù)等。為分布式框架提供協(xié)調(diào)服務(wù)的Apache項目。
?

Zookeeper 數(shù)據(jù)結(jié)構(gòu)：?

ZooKeeper數(shù)據(jù)模型的結(jié)構(gòu)與Linux文件系統(tǒng)很類似，整體上可以看作是一棵樹，每個節(jié)點(diǎn)稱做一個ZNode。每一個ZNode默認(rèn)能夠存儲1MB的數(shù)據(jù)，每個ZNode都可以通過其路徑唯一標(biāo)識。

1.2 ?Zookeeper 工作機(jī)制

?Zookeeper從設(shè)計模式角度來理解：是一個基于觀察者模式設(shè)計的分布式服務(wù)管理框架，它負(fù)責(zé)存儲和管理大家都關(guān)心的數(shù)據(jù)，然后接受觀察者的注冊，一旦這些數(shù)據(jù)的狀態(tài)發(fā)生變化，Zookeeper就將負(fù)責(zé)通知已經(jīng)在Zookeeper上注冊的那些觀察者做出相應(yīng)的反應(yīng)。也就是說 Zookeeper = 文件系統(tǒng) + 通知機(jī)制。
?

?1.4 ?Zookeeper 應(yīng)用場景

提供的服務(wù)包括：統(tǒng)一命名服務(wù)、統(tǒng)一配置管理、統(tǒng)一集群管理、服務(wù)器節(jié)點(diǎn)動態(tài)上下線、軟負(fù)載均衡等。
●統(tǒng)一命名服務(wù)
在分布式環(huán)境下，經(jīng)常需要對應(yīng)用/服務(wù)進(jìn)行統(tǒng)一命名，便于識別。例如：IP不容易記住，而域名容易記住。

●統(tǒng)一配置管理
（1）分布式環(huán)境下，配置文件同步非常常見。一般要求一個集群中，所有節(jié)點(diǎn)的配置信息是一致的，比如Kafka集群。對配置文件修改后，希望能夠快速同步到各個節(jié)點(diǎn)上。
（2）配置管理可交由ZooKeeper實(shí)現(xiàn)?？蓪⑴渲眯畔懭隯ooKeeper上的一個Znode。各個客戶端服務(wù)器監(jiān)聽這個Znode。一旦 Znode中的數(shù)據(jù)被修改，ZooKeeper將通知各個客戶端服務(wù)器。

●統(tǒng)一集群管理
（1）分布式環(huán)境中，實(shí)時掌握每個節(jié)點(diǎn)的狀態(tài)是必要的。可根據(jù)節(jié)點(diǎn)實(shí)時狀態(tài)做出一些調(diào)整。
（2）ZooKeeper可以實(shí)現(xiàn)實(shí)時監(jiān)控節(jié)點(diǎn)狀態(tài)變化。可將節(jié)點(diǎn)信息寫入ZooKeeper上的一個ZNode。監(jiān)聽這個ZNode可獲取它的實(shí)時狀態(tài)變化。

●服務(wù)器動態(tài)上下線
客戶端能實(shí)時洞察到服務(wù)器上下線的變化。

●軟負(fù)載均衡
在Zookeeper中記錄每臺服務(wù)器的訪問數(shù)，讓訪問數(shù)最少的服務(wù)器去處理最新的客戶端請求。

1.5?Zookeeper 選舉機(jī)制??

第一次啟動選舉機(jī)制

（1）服務(wù)器1啟動，發(fā)起一次選舉。服務(wù)器1投自己一票。此時服務(wù)器1票數(shù)一票，不夠半數(shù)以上（3票），選舉無法完成，服務(wù)器1狀態(tài)保持為LOOKING；

（2）服務(wù)器2啟動，再發(fā)起一次選舉。服務(wù)器1和2分別投自己一票并交換選票信息：此時服務(wù)器1發(fā)現(xiàn)服務(wù)器2的myid比自己目前投票推舉的（服務(wù)器1）大，更改選票為推舉服務(wù)器2。此時服務(wù)器1票數(shù)0票，服務(wù)器2票數(shù)2票，沒有半數(shù)以上結(jié)果，選舉無法完成，服務(wù)器1，2狀態(tài)保持LOOKING

（3）服務(wù)器3啟動，發(fā)起一次選舉。此時服務(wù)器1和2都會更改選票為服務(wù)器3。此次投票結(jié)果：服務(wù)器1為0票，服務(wù)器2為0票，服務(wù)器3為3票。此時服務(wù)器3的票數(shù)已經(jīng)超過半數(shù)，服務(wù)器3當(dāng)選Leader。服務(wù)器1，2更改狀態(tài)為FOLLOWING，服務(wù)器3更改狀態(tài)為LEADING；

（4）服務(wù)器4啟動，發(fā)起一次選舉。此時服務(wù)器1，2，3已經(jīng)不是LOOKING狀態(tài)，不會更改選票信息。交換選票信息結(jié)果：服務(wù)器3為3票，服務(wù)器4為1票。此時服務(wù)器4服從多數(shù)，更改選票信息為服務(wù)器3，并更改狀態(tài)為FOLLOWING；（5）服務(wù)器5啟動，同4一樣當(dāng)小弟。
?

?文章來源地址http://www.zghlxwxcb.cn/news/detail-464517.html

非第一次啟動選舉機(jī)制

（1）當(dāng)ZooKeeper 集群中的一臺服務(wù)器出現(xiàn)以下兩種情況之一時，就會開始進(jìn)入Leader選舉：

1）服務(wù)器初始化啟動。

2）服務(wù)器運(yùn)行期間無法和Leader保持連接.

（2）而當(dāng)一臺機(jī)器進(jìn)入Leader選舉流程時，當(dāng)前集群也可能會處于以下兩種狀態(tài)：
?

1）集群中本來就已經(jīng)存在一個Leader。
對于已經(jīng)存在Leader的情況，機(jī)器試圖去選舉Leader時，會被告知當(dāng)前服務(wù)器的Leader信息，對于該機(jī)器來說，僅僅需要和 Leader機(jī)器建立連接，并進(jìn)行狀態(tài)同步即可。

2）集群中確實(shí)不存在Leader。
假設(shè)ZooKeeper由5臺服務(wù)器組成，SID分別為1、2、3、4、5，ZXID分別為8、8、8、7、7，并且此時SID為3的服務(wù)器是Leader。某一時刻，3和5服務(wù)器出現(xiàn)故障，因此開始進(jìn)行Leader選舉。
選舉Leader規(guī)則：
1.EPOCH大的直接勝出
2.EPOCH相同，事務(wù)id大的勝出
3.事務(wù)id相同，服務(wù)器id大的勝出
?

SID：服務(wù)器ID。用來唯一標(biāo)識一臺ZooKeeper集群中的機(jī)器，每臺機(jī)器不能重復(fù)，和myid一致。
ZXID：事務(wù)ID。ZXID是一個事務(wù)ID，用來標(biāo)識一次服務(wù)器狀態(tài)的變更。在某一時刻，集群中的每臺機(jī)器的ZXID值不一定完全一致，這和ZooKeeper服務(wù)器對于客戶端“更新請求”的處理邏輯速度有關(guān)。
Epoch：每個Leader任期的代號。沒有Leader時同一輪投票過程中的邏輯時鐘值是相同的。每投完一次票這個數(shù)據(jù)就會增加

2. Zookeeper 集群的部署?

?

步驟一：zookeeper服務(wù)安裝前的準(zhǔn)備?

對三臺服務(wù)器均操作執(zhí)行 ：

//關(guān)閉防火墻
systemctl stop firewalld
systemctl disable firewalld
setenforce 0
?
//安裝 JDK
yum install -y java-1.8.0-openjdk java-1.8.0-openjdk-devel
java -version
?
//下載安裝包
官方下載地址：https://archive.apache.org/dist/zookeeper/
?
cd /opt
wget https://archive.apache.org/dist/zookeeper/zookeeper-3.5.7/apache-zookeeper-3.5.7-bin.tar.gz

步驟二：安裝 Zookeeper?服務(wù)

cd /opt
tar -zxvf apache-zookeeper-3.5.7-bin.tar.gz
mv apache-zookeeper-3.5.7-bin /usr/local/zookeeper-3.5.7
?
//修改配置文件
cd /usr/local/zookeeper-3.5.7/conf/
cp zoo_sample.cfg zoo.cfg
?
vim zoo.cfg
tickTime=2000 ? #通信心跳時間，Zookeeper服務(wù)器與客戶端心跳時間，單位毫秒
initLimit=10 ? ?#Leader和Follower初始連接時能容忍的最多心跳數(shù)（tickTime的數(shù)量），這里表示為10*2s
syncLimit=5 ? ? #Leader和Follower之間同步通信的超時時間，這里表示如果超過5*2s，Leader認(rèn)為Follwer死掉，并從服務(wù)器列表中刪除Follwer
dataDir=/usr/local/zookeeper-3.5.7/data ? ? ?●修改，指定保存Zookeeper中的數(shù)據(jù)的目錄，目錄需要單獨(dú)創(chuàng)建
dataLogDir=/usr/local/zookeeper-3.5.7/logs ? ●添加，指定存放日志的目錄，目錄需要單獨(dú)創(chuàng)建
clientPort=2181 ? #客戶端連接端口
#添加集群信息
server.1=192.168.50.24:3188:3288
server.2=192.168.50.25:3188:3288
server.3=192.168.50.26:3188:3288
?
-------------------------------------------------------------------------------------
server.A=B:C:D
●A是一個數(shù)字，表示這個是第幾號服務(wù)器。集群模式下需要在zoo.cfg中dataDir指定的目錄下創(chuàng)建一個文件myid，這個文件里面有一個數(shù)據(jù)就是A的值，Zookeeper啟動時讀取此文件，拿到里面的數(shù)據(jù)與zoo.cfg里面的配置信息比較從而判斷到底是哪個server。
●B是這個服務(wù)器的地址。
●C是這個服務(wù)器Follower與集群中的Leader服務(wù)器交換信息的端口。
●D是萬一集群中的Leader服務(wù)器掛了，需要一個端口來重新進(jìn)行選舉，選出一個新的Leader，而這個端口就是用來執(zhí)行選舉時服務(wù)器相互通信的端口。
-------------------------------------------------------------------------------------
?
//拷貝配置好的 Zookeeper 配置文件到其他機(jī)器上
scp /usr/local/zookeeper-3.5.7/conf/zoo.cfg 192.168.73.110:/usr/local/zookeeper-3.5.7/conf/
scp /usr/local/zookeeper-3.5.7/conf/zoo.cfg 192.168.73.111:/usr/local/zookeeper-3.5.7/conf/
?
//在每個節(jié)點(diǎn)上創(chuàng)建數(shù)據(jù)目錄和日志目錄
mkdir /usr/local/zookeeper-3.5.7/data
mkdir /usr/local/zookeeper-3.5.7/logs
?
//在每個節(jié)點(diǎn)的dataDir指定的目錄下創(chuàng)建一個 myid 的文件
echo 1 > /usr/local/zookeeper-3.5.7/data/myid
echo 2 > /usr/local/zookeeper-3.5.7/data/myid
echo 3 > /usr/local/zookeeper-3.5.7/data/myid

?步驟三：配置 Zookeeper 啟動腳本

vim /etc/init.d/zookeeper
#!/bin/bash
#chkconfig:2345 20 90
#description:Zookeeper Service Control Script
ZK_HOME='/usr/local/zookeeper-3.5.7'
case $1 in
start)
?? ?echo "---------- zookeeper 啟動 ------------"
?? ?$ZK_HOME/bin/zkServer.sh start
;;
stop)
?? ?echo "---------- zookeeper 停止 ------------"
?? ?$ZK_HOME/bin/zkServer.sh stop
;;
restart)
?? ?echo "---------- zookeeper 重啟 ------------"
?? ?$ZK_HOME/bin/zkServer.sh restart
;;
status)
?? ?echo "---------- zookeeper 狀態(tài) ------------"
?? ?$ZK_HOME/bin/zkServer.sh status
;;
*)
? ? echo "Usage: $0 {start|stop|restart|status}"
esac
###將腳本轉(zhuǎn)到其他兩臺服務(wù)器上
cd /etc/init.d/
scp zookeeper root@192.168.50.25:`pwd`
scp zookeeper root@192.168.50.26:`pwd`
?
//?? ?設(shè)置開機(jī)自啟
chmod +x /etc/init.d/zookeeper
chkconfig --add zookeeper

?

3.消息隊列的相關(guān)知識

3.1 消息隊列的概述?

需要消息隊列（MQ）的原因：

主要原因是由于在高并發(fā)環(huán)境下，同步請求來不及處理，請求往往會發(fā)生阻塞。比如大量的請求并發(fā)訪問數(shù)據(jù)庫，導(dǎo)致行鎖表鎖，最后請求線程會堆積過多，從而觸發(fā) too many connection 錯誤，引發(fā)雪崩效應(yīng)。
我們使用消息隊列，通過異步處理請求，從而緩解系統(tǒng)的壓力。消息隊列常應(yīng)用于異步處理，流量削峰，應(yīng)用解耦，消息通訊等場景。

當(dāng)前比較常見的 MQ 中間件有 ActiveMQ、RabbitMQ、RocketMQ、Kafka 等。
?

3.2 消息隊列的優(yōu)點(diǎn)?

（1）解耦

允許你獨(dú)立的擴(kuò)展或修改兩邊的處理過程，只要確保它們遵守同樣的接口約束。

（2）可恢復(fù)性

?系統(tǒng)的一部分組件失效時，不會影響到整個系統(tǒng)。消息隊列降低了進(jìn)程間的耦合度，所以即使一個處理消息的進(jìn)程掛掉，加入隊列中的消息仍然可以在系統(tǒng)恢復(fù)后被處理。

（3）緩沖

?

?有助于控制和優(yōu)化數(shù)據(jù)流經(jīng)過系統(tǒng)的速度，解決生產(chǎn)消息和消費(fèi)消息的處理速度不一致的情況。

（4）靈活性 & 峰值處理能力

在訪問量劇增的情況下，應(yīng)用仍然需要繼續(xù)發(fā)揮作用，但是這樣的突發(fā)流量并不常見。如果為以能處理這類峰值訪問為標(biāo)準(zhǔn)來投入資源隨時待命無疑是巨大的浪費(fèi)。使用消息隊列能夠使關(guān)鍵組件頂住突發(fā)的訪問壓力，而不會因?yàn)橥话l(fā)的超負(fù)荷的請求而完全崩潰。

（5）異步通信

很多時候，用戶不想也不需要立即處理消息。消息隊列提供了異步處理機(jī)制，允許用戶把一個消息放入隊列，但并不立即處理它。想向隊列中放入多少消息就放多少，然后在需要的時候再去處理它們。
?

3.3 消息隊列的兩種模式

（1）點(diǎn)對點(diǎn)模式（一對一，消費(fèi)者主動拉取數(shù)據(jù)，消息收到后消息清除）

消息生產(chǎn)者生產(chǎn)消息發(fā)送到消息隊列中，然后消息消費(fèi)者從消息隊列中取出并且消費(fèi)消息。消息被消費(fèi)以后，消息隊列中不再有存儲，所以消息消費(fèi)者不可能消費(fèi)到已經(jīng)被消費(fèi)的消息。消息隊列支持存在多個消費(fèi)者，但是對一個消息而言，只會有一個消費(fèi)者可以消費(fèi)。

（2）發(fā)布/訂閱模式（一對多，又叫觀察者模式，消費(fèi)者消費(fèi)數(shù)據(jù)之后不會清除消息）

消息生產(chǎn)者（發(fā)布）將消息發(fā)布到 topic 中，同時有多個消息消費(fèi)者（訂閱）消費(fèi)該消息。和點(diǎn)對點(diǎn)方式不同，發(fā)布到 topic 的消息會被所有訂閱者消費(fèi)。
發(fā)布/訂閱模式是定義對象間一種一對多的依賴關(guān)系，使得每當(dāng)一個對象（目標(biāo)對象）的狀態(tài)發(fā)生改變，則所有依賴于它的對象（觀察者對象）都會得到通知并自動更新。

??4.Kafka消息隊列工具的相關(guān)知識

?

?

?4.1 Kafka的定義與簡介

Kafka 是一個分布式的基于發(fā)布/訂閱模式的消息隊列（MQ，Message Queue），主要應(yīng)用于大數(shù)據(jù)實(shí)時處理領(lǐng)域。?

Kafka 是最初由 Linkedin 公司開發(fā)，是一個分布式、支持分區(qū)的（partition）、多副本的（replica），基于 Zookeeper 協(xié)調(diào)的分布式消息中間件系統(tǒng)，它的最大的特性就是可以實(shí)時的處理大量數(shù)據(jù)以滿足各種需求場景，比如基于 hadoop 的批處理系統(tǒng)、低延遲的實(shí)時系統(tǒng)、Spark/Flink 流式處理引擎，nginx 訪問日志，消息服務(wù)等等，用 scala 語言編寫，
Linkedin 于 2010 年貢獻(xiàn)給了 Apache 基金會并成為頂級開源項目。
?

4.2 Kafka的特性?

●高吞吐量、低延遲
Kafka 每秒可以處理幾十萬條消息，它的延遲最低只有幾毫秒。每個 topic 可以分多個 Partition，Consumer Group 對 Partition 進(jìn)行消費(fèi)操作，提高負(fù)載均衡能力和消費(fèi)能力。

●可擴(kuò)展性
kafka 集群支持熱擴(kuò)展

●持久性、可靠性
消息被持久化到本地磁盤，并且支持?jǐn)?shù)據(jù)備份防止數(shù)據(jù)丟失

●容錯性
允許集群中節(jié)點(diǎn)失?。ǘ喔北厩闆r下，若副本數(shù)量為 n，則允許 n-1 個節(jié)點(diǎn)失敗）

●高并發(fā)
支持?jǐn)?shù)千個客戶端同時讀寫
?

4.3 Kafka 系統(tǒng)架構(gòu)?

?（1）Broker
一臺 kafka 服務(wù)器就是一個 broker。一個集群由多個 broker 組成。一個 broker 可以容納多個 topic。

（2）Topic
?可以理解為一個隊列，生產(chǎn)者和消費(fèi)者面向的都是一個 topic。
類似于數(shù)據(jù)庫的表名或者 ES 的 index
物理上不同 topic 的消息分開存儲

（3）Partition
?為了實(shí)現(xiàn)擴(kuò)展性，一個非常大的 topic 可以分布到多個 broker（即服務(wù)器）上，一個 topic 可以分割為一個或多個 partition，每個 partition 是一個有序的隊列。Kafka 只保證 partition 內(nèi)的記錄是有序的，而不保證 topic 中不同 partition 的順序。
?

（4）Replica
?副本，為保證集群中的某個節(jié)點(diǎn)發(fā)生故障時，該節(jié)點(diǎn)上的 partition 數(shù)據(jù)不丟失，且 kafka 仍然能夠繼續(xù)工作，kafka 提供了副本機(jī)制，一個 topic 的每個分區(qū)都有若干個副本，一個 leader 和若干個 follower。

（5）Leader
?每個 partition 有多個副本，其中有且僅有一個作為 Leader，Leader 是當(dāng)前負(fù)責(zé)數(shù)據(jù)的讀寫的 partition。

（6）Follower
?Follower 跟隨 Leader，所有寫請求都通過 Leader 路由，數(shù)據(jù)變更會廣播給所有 Follower，F(xiàn)ollower 與 Leader 保持?jǐn)?shù)據(jù)同步。Follower 只負(fù)責(zé)備份，不負(fù)責(zé)數(shù)據(jù)的讀寫。
如果 Leader 故障，則從 Follower 中選舉出一個新的 Leader。
當(dāng) Follower 掛掉、卡住或者同步太慢，Leader 會把這個 Follower 從 ISR（Leader 維護(hù)的一個和 Leader 保持同步的 Follower 集合） 列表中刪除，重新創(chuàng)建一個 Follower。

（7）Producer
?生產(chǎn)者即數(shù)據(jù)的發(fā)布者，該角色將消息 push 發(fā)布到 Kafka 的 topic 中。
broker 接收到生產(chǎn)者發(fā)送的消息后，broker 將該消息追加到當(dāng)前用于追加數(shù)據(jù)的 segment 文件中。
生產(chǎn)者發(fā)送的消息，存儲到一個 partition 中，生產(chǎn)者也可以指定數(shù)據(jù)存儲的 partition。

（8）Consumer
?消費(fèi)者可以從 broker 中 pull 拉取數(shù)據(jù)。消費(fèi)者可以消費(fèi)多個 topic 中的數(shù)據(jù)。

（9）Consumer Group（CG）
?消費(fèi)者組，由多個 consumer 組成。
所有的消費(fèi)者都屬于某個消費(fèi)者組，即消費(fèi)者組是邏輯上的一個訂閱者?？蔀槊總€消費(fèi)者指定組名，若不指定組名則屬于默認(rèn)的組。
將多個消費(fèi)者集中到一起去處理某一個 Topic 的數(shù)據(jù)，可以更快的提高數(shù)據(jù)的消費(fèi)能力。
消費(fèi)者組內(nèi)每個消費(fèi)者負(fù)責(zé)消費(fèi)不同分區(qū)的數(shù)據(jù)，一個分區(qū)只能由一個組內(nèi)消費(fèi)者消費(fèi)，防止數(shù)據(jù)被重復(fù)讀取。
消費(fèi)者組之間互不影響。

（10）offset 偏移量
可以唯一的標(biāo)識一條消息。
偏移量決定讀取數(shù)據(jù)的位置，不會有線程安全的問題，消費(fèi)者通過偏移量來決定下次讀取的消息（即消費(fèi)位置）。
消息被消費(fèi)之后，并不被馬上刪除，這樣多個業(yè)務(wù)就可以重復(fù)使用 Kafka 的消息。
某一個業(yè)務(wù)也可以通過修改偏移量達(dá)到重新讀取消息的目的，偏移量由用戶控制。
消息最終還是會被刪除的，默認(rèn)生命周期為 1 周（7*24小時）。

（11）Zookeeper
?Kafka 通過 Zookeeper 來存儲集群的 meta 信息。

由于 consumer 在消費(fèi)過程中可能會出現(xiàn)斷電宕機(jī)等故障，consumer 恢復(fù)后，需要從故障前的位置的繼續(xù)消費(fèi)，所以 consumer 需要實(shí)時記錄自己消費(fèi)到了哪個 offset，以便故障恢復(fù)后繼續(xù)消費(fèi)。
Kafka 0.9 版本之前，consumer 默認(rèn)將 offset 保存在 Zookeeper 中；從 0.9 版本開始，consumer 默認(rèn)將 offset 保存在 Kafka 一個內(nèi)置的 topic 中，該 topic 為 __consumer_offsets。

也就是說，zookeeper的作用就是，生產(chǎn)者push數(shù)據(jù)到kafka集群，就必須要找到kafka集群的節(jié)點(diǎn)在哪里，這些都是通過zookeeper去尋找的。消費(fèi)者消費(fèi)哪一條數(shù)據(jù)，也需要zookeeper的支持，從zookeeper獲得offset，offset記錄上一次消費(fèi)的數(shù)據(jù)消費(fèi)到哪里，這樣就可以接著下一條數(shù)據(jù)進(jìn)行消費(fèi)。

?

?

每個 topic 至少有一個 partition，當(dāng)生產(chǎn)者產(chǎn)生數(shù)據(jù)的時候，會根據(jù)分配策略選擇分區(qū)，然后將消息追加到指定的分區(qū)的隊列末尾。
##Partation 數(shù)據(jù)路由規(guī)則：
1．指定了 patition，則直接使用；
2．未指定 patition 但指定 key（相當(dāng)于消息中某個屬性），通過對 key 的 value 進(jìn)行 hash 取模，選出一個 patition；
3．patition 和 key 都未指定，使用輪詢選出一個 patition。
?
每條消息都會有一個自增的編號，用于標(biāo)識消息的偏移量，標(biāo)識順序從 0 開始。
?
每個 partition 中的數(shù)據(jù)使用多個 segment 文件存儲。
?
如果 topic 有多個 partition，消費(fèi)數(shù)據(jù)時就不能保證數(shù)據(jù)的順序。嚴(yán)格保證消息的消費(fèi)順序的場景下（例如商品秒殺、 搶紅包），需要將 partition 數(shù)目設(shè)為 1。
?
●broker 存儲 topic 的數(shù)據(jù)。如果某 topic 有 N 個 partition，集群有 N 個 broker，那么每個 broker 存儲該 topic 的一個 partition。
●如果某 topic 有 N 個 partition，集群有 (N+M) 個 broker，那么其中有 N 個 broker 存儲 topic 的一個 partition， 剩下的 M 個 broker 不存儲該 topic 的 partition 數(shù)據(jù)。
●如果某 topic 有 N 個 partition，集群中 broker 數(shù)目少于 N 個，那么一個 broker 存儲該 topic 的一個或多個 partition。在實(shí)際生產(chǎn)環(huán)境中，盡量避免這種情況的發(fā)生，這種情況容易導(dǎo)致 Kafka 集群數(shù)據(jù)不均衡。

5.? ?部署 kafka 集群

?pache公司的軟件包官方下載地址：archive.apache.org/dist/

注：kafka從3.0版本之后，不再依賴zookeeper。但是為了業(yè)務(wù)的穩(wěn)定，大部分企業(yè)還是采用zookeeper+Kafka的架構(gòu)

組件部署

?

?具體部署的步驟

步驟一：安裝的包的下載?

1.下載安裝包
官方下載地址：http://kafka.apache.org/downloads.html
?
cd /opt
wget https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.7.1/kafka_2.13-2.7.1.tgz
?
?
2.安裝 Kafka
cd /opt/
tar zxvf kafka_2.13-2.7.1.tgz
mv kafka_2.13-2.7.1 /usr/local/kafka

步驟二：修改Kafka的配置文件?

cd /usr/local/kafka/config/
cp server.properties{,.bak}
?
vim server.properties
broker.id=0 ? ?●21行，broker的全局唯一編號，每個broker不能重復(fù)，因此要在其他機(jī)器上配置 broker.id=1、broker.id=2
listeners=PLAINTEXT://192.168.50.24:9092 ? ?●31行，指定監(jiān)聽的IP和端口，如果修改每個broker的IP需區(qū)分開來，也可保持默認(rèn)配置不用修改
num.network.threads=3 ? ?#42行，broker 處理網(wǎng)絡(luò)請求的線程數(shù)量，一般情況下不需要去修改
num.io.threads=8 ? ? ? ? #45行，用來處理磁盤IO的線程數(shù)量，數(shù)值應(yīng)該大于硬盤數(shù)
socket.send.buffer.bytes=102400 ? ? ? #48行，發(fā)送套接字的緩沖區(qū)大小
socket.receive.buffer.bytes=102400 ? ?#51行，接收套接字的緩沖區(qū)大小
socket.request.max.bytes=104857600 ? ?#54行，請求套接字的緩沖區(qū)大小
log.dirs=/usr/local/kafka/logs ? ? ? ?#60行，kafka運(yùn)行日志存放的路徑，也是數(shù)據(jù)存放的路徑
num.partitions=1 ? ?#65行，topic在當(dāng)前broker上的默認(rèn)分區(qū)個數(shù)，會被topic創(chuàng)建時的指定參數(shù)覆蓋
num.recovery.threads.per.data.dir=1 ? ?#69行，用來恢復(fù)和清理data下數(shù)據(jù)的線程數(shù)量
log.retention.hours=168 ? ?#103行，segment文件（數(shù)據(jù)文件）保留的最長時間，單位為小時，默認(rèn)為7天，超時將被刪除
log.segment.bytes=1073741824 ? ?#110行，一個segment文件最大的大小，默認(rèn)為 1G，超出將新建一個新的segment文件
zookeeper.connect=192.168.50.24:2181,192.168.50.25:2181,192.168.50.26:2181 ? ?●123行，配置連接Zookeeper集群地址
?
//修改環(huán)境變量
vim /etc/profile
export KAFKA_HOME=/usr/local/kafka
export PATH=$PATH:$KAFKA_HOME/bin
?
?
source /etc/profile

步驟三：將kafka添加到系統(tǒng)服務(wù)中?

//配置 kafka 啟動腳本
vim /etc/init.d/kafka
#!/bin/bash
#chkconfig:2345 22 88
#description:Kafka Service Control Script
KAFKA_HOME='/usr/local/kafka'
case $1 in
start)
?? ?echo "---------- Kafka 啟動 ------------"
?? ?${KAFKA_HOME}/bin/kafka-server-start.sh -daemon ${KAFKA_HOME}/config/server.properties
;;
stop)
?? ?echo "---------- Kafka 停止 ------------"
?? ?${KAFKA_HOME}/bin/kafka-server-stop.sh
;;
restart)
?? ?$0 stop
?? ?$0 start
;;
status)
?? ?echo "---------- Kafka 狀態(tài) ------------"
?? ?count=$(ps -ef | grep kafka | egrep -cv "grep|$$")
?? ?if [ "$count" -eq 0 ];then
? ? ? ? echo "kafka is not running"
? ? else
? ? ? ? echo "kafka is running"
? ? fi
;;
*)
? ? echo "Usage: $0 {start|stop|restart|status}"
esac
?
//設(shè)置開機(jī)自啟
cd /etc/init.d/
scp kafka root@192.168.50.25:`pwd`
scp kafka root@192.168.50.26:`pwd`
chmod +x /etc/init.d/kafka
chkconfig --add kafka
//分別啟動 Kafka
service kafka start

步驟四：進(jìn)行topic的管理?

3.Kafka 命令行操作
cd /usr/local/kafka/bin
//創(chuàng)建topic
kafka-topics.sh --create --zookeeper 192.168.50.24:2181,192.168.50.25:2181,192.168.50.26:2181 --replication-factor 2 --partitions 3 --topic test
?
-------------------------------------------------------------------------------------
--zookeeper：定義 zookeeper 集群服務(wù)器地址，如果有多個 IP 地址使用逗號分割，一般使用一個 IP 即可
：定義分區(qū)副本數(shù)，1 代表單副本，建議為 2?
--partitions：定義分區(qū)數(shù)?
--topic：定義 topic 名稱
-------------------------------------------------------------------------------------
?
//查看當(dāng)前服務(wù)器中的所有 topic
kafka-topics.sh --list --zookeeper 192.168.50.25:2181,192.168.50.26:2181,192.168.50.24:2181
?
//查看某個 topic 的詳情
kafka-topics.sh ?--describe --zookeeper 192.168.50.24:2181,192.168.50.25:2181,192.168.50.26:2181
?
//發(fā)布消息
kafka-console-producer.sh --broker-list 192.168.50.24:9092,192.168.50.25:9092,192.168.50.26:9092 ?--topic test
?
//消費(fèi)消息
kafka-console-consumer.sh --bootstrap-server 192.168.50.24:9092,192.168.50.25:9092,192.168.50.26:9092 --topic test --from-beginning
?
-------------------------------------------------------------------------------------
--from-beginning：會把主題中以往所有的數(shù)據(jù)都讀取出來
-------------------------------------------------------------------------------------
?
//修改分區(qū)數(shù)
kafka-topics.sh --zookeeper 192.168.50.24:2181,192.168.50.25:2181,192.168.50.26:2181 --alter --topic test --partitions 6
?
//刪除 topic
kafka-topics.sh --delete --zookeeper 192.168.50.24:2181,192.168.50.25:2181,192.50.26.111:2181 --topic test

?

?

?

?

//修改分區(qū)數(shù)
kafka-topics.sh --zookeeper 192.168.50.24:2181,192.168.50.25:2181,192.168.50.26:2181 --alter --topic test --partitions 6

//刪除 topic
kafka-topics.sh --delete --zookeeper 192.168.50.24:2181,192.168.50.25:2181,192.168.50.26:2181 --topic test

?

到了這里，關(guān)于分布式應(yīng)用之zookeeper集群+消息隊列Kafka的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！