一、Zookeeper
1.1、Zookeeper 定義
Zookeeper是一個(gè)開源的分布式的，為分布式框架提供協(xié)調(diào)服務(wù)的Apache項(xiàng)目。
1.2、Zookeeper 工作機(jī)制
Zookeeper從設(shè)計(jì)模式角度來理解：是一個(gè)基于觀察者模式設(shè)計(jì)的分布式服務(wù)管理框架，它負(fù)責(zé)存儲(chǔ)和管理大家都關(guān)心的數(shù)據(jù)，然后接受觀察者的注冊(cè)，一旦這些數(shù)據(jù)的狀態(tài)發(fā)生變化，Zookeeper就將負(fù)責(zé)通知已經(jīng)在Zookeeper上注冊(cè)的那些觀察者做出相應(yīng)的反應(yīng)。也就是說 Zookeeper = 文件系統(tǒng) + 通知機(jī)制。
1.3、Zookeeper 特點(diǎn)
- Zookeeper：一個(gè)領(lǐng)導(dǎo)者（Leader），多個(gè)跟隨者（Follower）組成的集群。
- Zookeeper集群中只要有半數(shù)以上節(jié)點(diǎn)存活，Zookeeper集群就能正常服務(wù)。所以Zookeeper適合安裝奇數(shù)臺(tái)服務(wù)器。
- 全局?jǐn)?shù)據(jù)一致：每個(gè)Server保存一份相同的數(shù)據(jù)副本，Client無論連接到哪個(gè)Server，數(shù)據(jù)都是一致的。
- 更新請(qǐng)求順序執(zhí)行，來自同一個(gè)Client的更新請(qǐng)求按其發(fā)送順序依次執(zhí)行，即先進(jìn)先出。
- 數(shù)據(jù)更新原子性，一次數(shù)據(jù)更新要么成功，要么失敗。
- 實(shí)時(shí)性，在一定時(shí)間范圍內(nèi)，Client能讀到最新數(shù)據(jù)。
1.4、Zookeeper 數(shù)據(jù)結(jié)構(gòu)
ZooKeeper數(shù)據(jù)模型的結(jié)構(gòu)與Linux文件系統(tǒng)很類似，整體上可以看作是一棵樹，每個(gè)節(jié)點(diǎn)稱做一個(gè)ZNode。每一個(gè)ZNode默認(rèn)能夠存儲(chǔ)1MB的數(shù)據(jù)，每個(gè)ZNode都可以通過其路徑唯一標(biāo)識(shí)。
1.5、Zookeeper 應(yīng)用場(chǎng)景
提供的服務(wù)包括：統(tǒng)一命名服務(wù)、統(tǒng)一配置管理、統(tǒng)一集群管理、服務(wù)器節(jié)點(diǎn)動(dòng)態(tài)上下線、軟負(fù)載均衡等。
-
統(tǒng)一命名服務(wù)
在分布式環(huán)境下，經(jīng)常需要對(duì)應(yīng)用/服務(wù)進(jìn)行統(tǒng)一命名，便于識(shí)別。例如：IP不容易記住，而域名容易記住。 -
統(tǒng)一配置管理
- 分布式環(huán)境下，配置文件同步非常常見。一般要求一個(gè)集群中，所有節(jié)點(diǎn)的配置信息是一致的，比如Kafka集群。對(duì)配置文件修改后，希望能夠快速同步到各個(gè)節(jié)點(diǎn)上。
- 配置管理可交由ZooKeeper實(shí)現(xiàn)?？蓪⑴渲眯畔懭隯ooKeeper上的一個(gè)Znode。各個(gè)客戶端服務(wù)器監(jiān)聽這個(gè)Znode。一旦Znode中的數(shù)據(jù)被修改，ZooKeeper將通知各個(gè)客戶端服務(wù)器。
-
統(tǒng)一集群管理
- 分布式環(huán)境中，實(shí)時(shí)掌握每個(gè)節(jié)點(diǎn)的狀態(tài)是必要的。可根據(jù)節(jié)點(diǎn)實(shí)時(shí)狀態(tài)做出一些調(diào)整。
- ZooKeeper可以實(shí)現(xiàn)實(shí)時(shí)監(jiān)控節(jié)點(diǎn)狀態(tài)變化?？蓪⒐?jié)點(diǎn)信息寫入ZooKeeper上的一個(gè)ZNode。監(jiān)聽這個(gè)ZNode可獲取它的實(shí)時(shí)狀態(tài)變化。
-
服務(wù)器動(dòng)態(tài)上下線
客戶端能實(shí)時(shí)洞察到服務(wù)器上下線的變化。 -
軟負(fù)載均衡
在Zookeeper中記錄每臺(tái)服務(wù)器的訪問數(shù)，讓訪問數(shù)最少的服務(wù)器去處理最新的客戶端請(qǐng)求。
1.5、Zookeeper 選舉機(jī)制
1.5.1、 第一次啟動(dòng)選舉機(jī)制
- 服務(wù)器1啟動(dòng)，發(fā)起一次選舉。服務(wù)器1投自己一票。此時(shí)服務(wù)器1票數(shù)一票，不夠半數(shù)以上（3票），選舉無法完成，服務(wù)器1狀態(tài)保持為L(zhǎng)OOKING；
- 服務(wù)器2啟動(dòng)，再發(fā)起一次選舉。服務(wù)器1和2分別投自己一票并交換選票信息：此時(shí)服務(wù)器1發(fā)現(xiàn)服務(wù)器2的myid比自己目前投票推舉的（服務(wù)器1）大，更改選票為推舉服務(wù)器2。此時(shí)服務(wù)器1票數(shù)0票，服務(wù)器2票數(shù)2票，沒有半數(shù)以上結(jié)果，選舉無法完成，服務(wù)器1，2狀態(tài)保持LOOKING
- 服務(wù)器3啟動(dòng)，發(fā)起一次選舉。此時(shí)服務(wù)器1和2都會(huì)更改選票為服務(wù)器3。此次投票結(jié)果：服務(wù)器1為0票，服務(wù)器2為0票，服務(wù)器3為3票。此時(shí)服務(wù)器3的票數(shù)已經(jīng)超過半數(shù)，服務(wù)器3當(dāng)選Leader。服務(wù)器1，2更改狀態(tài)為FOLLOWING，服務(wù)器3更改狀態(tài)為L(zhǎng)EADING；
- 服務(wù)器4啟動(dòng)，發(fā)起一次選舉。此時(shí)服務(wù)器1，2，3已經(jīng)不是LOOKING狀態(tài)，不會(huì)更改選票信息。交換選票信息結(jié)果：服務(wù)器3為3票，服務(wù)器4為1票。此時(shí)服務(wù)器4服從多數(shù)，更改選票信息為服務(wù)器3，并更改狀態(tài)為FOLLOWING；（5）服務(wù)器5啟動(dòng)，同4一樣當(dāng)小弟。
1.5.2、 非第一次啟動(dòng)選舉機(jī)制
-
當(dāng)ZooKeeper 集群中的一臺(tái)服務(wù)器出現(xiàn)以下兩種情況之一時(shí)，就會(huì)開始進(jìn)入Leader選舉：
- 服務(wù)器初始化啟動(dòng)。
- 服務(wù)器運(yùn)行期間無法和Leader保持連接。
-
而當(dāng)一臺(tái)機(jī)器進(jìn)入Leader選舉流程時(shí)，當(dāng)前集群也可能會(huì)處于以下兩種狀態(tài)：
- 集群中本來就已經(jīng)存在一個(gè)Leader。
- 對(duì)于已經(jīng)存在Leader的情況，機(jī)器試圖去選舉Leader時(shí)，會(huì)被告知當(dāng)前服務(wù)器的Leader信息，對(duì)于該機(jī)器來說，僅僅需要和 Leader機(jī)器建立連接，并進(jìn)行狀態(tài)同步即可。
-
集群中確實(shí)不存在Leader。
假設(shè)ZooKeeper由5臺(tái)服務(wù)器組成，SID分別為1、2、3、4、5，ZXID分別為8、8、8、7、7，并且此時(shí)SID為3的服務(wù)器是Leader。某一時(shí)刻，3和5服務(wù)器出現(xiàn)故障，因此開始進(jìn)行Leader選舉。選舉Leader規(guī)則：
1.EPOCH大的直接勝出
2.EPOCH相同，事務(wù)id大的勝出
3.事務(wù)id相同，服務(wù)器id大的勝出
二、Zookeeper集群部署
#準(zhǔn)備 3 臺(tái)服務(wù)器做 Zookeeper 集群
192.168.11.16
192.168.11.17
192.168.11.18
2.1、 安裝前準(zhǔn)備
#關(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.6.4/apache-zookeeper-3.6.4-bin.tar.gz
2.2、 安裝 Zookeeper
解壓代碼包
cd /opt
#解壓
tar -zxvf apache-zookeeper-3.6.4-bin.tar.gz
mv apache-zookeeper-3.6.4-bin /usr/local/zookeeper-3.6.4
修改配置文件
cd /usr/local/zookeeper-3.6.4/conf/
#配置模板
cp zoo_sample.cfg zoo.cfg
#修改配置文件
vim zoo.cfg
tickTime=2000
#通信心跳時(shí)間，Zookeeper服務(wù)器與客戶端心跳時(shí)間，單位毫秒
initLimit=10
#Leader和Follower初始連接時(shí)能容忍的最多心跳數(shù)（tickTime的數(shù)量），這里表示為10*2s
syncLimit=5
#Leader和Follower之間同步通信的超時(shí)時(shí)間，這里表示如果超過5*2s，Leader認(rèn)為Follwer死掉，并從服務(wù)器列表中刪除Follwer
dataDir=/usr/local/zookeeper-3.6.4/data
#修改，指定保存Zookeeper中的數(shù)據(jù)的目錄，目錄需要單獨(dú)創(chuàng)建
dataLogDir=/usr/local/zookeeper-3.6.4/logs
#添加，指定存放日志的目錄，目錄需要單獨(dú)創(chuàng)建
clientPort=2181 #客戶端連接端口
#添加集群信息
server.1=192.168.11.16:3188:3288
server.2=192.168.11.17:3188:3288
server.3=192.168.11.18:3188:3288
#補(bǔ)充知識(shí)
server.A=B:C:D
-A是一個(gè)數(shù)字，表示這個(gè)是第幾號(hào)服務(wù)器。集群模式下需要在zoo.cfg中dataDir指定的目錄下創(chuàng)建一個(gè)文件myid，這個(gè)文件里面有一個(gè)數(shù)據(jù)就是A的值，Zookeeper啟動(dòng)時(shí)讀取此文件，拿到里面的數(shù)據(jù)與zoo.cfg里面的配置信息比較從而判斷到底是哪個(gè)server。
-B是這個(gè)服務(wù)器的地址。
-C是這個(gè)服務(wù)器Follower與集群中的Leader服務(wù)器交換信息的端口。
-D是萬一集群中的Leader服務(wù)器掛了，需要一個(gè)端口來重新進(jìn)行選舉，選出一個(gè)新的Leader，而這個(gè)端口就是用來執(zhí)行選舉時(shí)服務(wù)器相互通信的端口。
創(chuàng)建相關(guān)目錄和文件
#拷貝配置好的 Zookeeper 配置文件到其他機(jī)器上
scp /usr/local/zookeeper-3.6.4/conf/zoo.cfg 192.168.11.17:/usr/local/zookeeper-3.6.4/conf/
scp /usr/local/zookeeper-3.6.4/conf/zoo.cfg 192.168.11.18:/usr/local/zookeeper-3.6.4/conf/
#在每個(gè)節(jié)點(diǎn)上創(chuàng)建數(shù)據(jù)目錄和日志目錄
mkdir /usr/local/zookeeper-3.6.4/data
mkdir /usr/local/zookeeper-3.6.4/logs
#在每個(gè)節(jié)點(diǎn)的dataDir指定的目錄下創(chuàng)建一個(gè) myid 的文件
echo 1 > /usr/local/zookeeper-3.6.4/data/myid
echo 2 > /usr/local/zookeeper-3.6.4/data/myid
echo 3 > /usr/local/zookeeper-3.6.4/data/myid
編寫 Zookeeper 啟動(dòng)腳本
#Zookeeper 啟動(dòng)腳本
vim /etc/init.d/zookeeper
#!/bin/bash
#chkconfig: 2345 20 90
#description:Zookeeper Service Control Script
ZK_HOME='/usr/local/zookeeper-3.6.4'
case $1 in
start)
echo "---------- zookeeper 啟動(dòng) ------------"
$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
啟動(dòng)ZK，查看狀態(tài)
#設(shè)置開機(jī)自啟
chmod +x /etc/init.d/zookeeper
chkconfig --add zookeeper
#分別啟動(dòng) Zookeeper
service zookeeper start
#查看當(dāng)前狀態(tài)
service zookeeper status
三、消息隊(duì)列
3.1、 為什么要用消息隊(duì)列？
-
主要原因是由于在高并發(fā)環(huán)境下，同步請(qǐng)求來不及處理，請(qǐng)求往往會(huì)發(fā)生阻塞。
-
比如大量的請(qǐng)求并發(fā)訪問數(shù)據(jù)庫(kù)，導(dǎo)致行鎖表鎖，最后請(qǐng)求線程會(huì)堆積過多，從而觸發(fā) too many connection 錯(cuò)誤，引發(fā)雪崩效應(yīng)。
-
使用消息隊(duì)列，通過異步處理請(qǐng)求，從而緩解系統(tǒng)的壓力。
3.2、 中間件
-
中間件用于實(shí)現(xiàn)應(yīng)用解耦、異步處理。
-
web應(yīng)用型（代理服務(wù)器）：Nginx 、Haproxy 、Tomcat PHP
-
消息隊(duì)列型（MQ):Active MQ 、RabbitMQ 、RocketMQ 、Kafka
3.3、 使用消息隊(duì)列的好處
-
應(yīng)用解耦
- 允許你獨(dú)立的擴(kuò)展或修改兩邊的處理過程，只要確保它們遵守同樣的接口約束。
-
可恢復(fù)性
- 消息隊(duì)列降低了進(jìn)程間的耦合度，所以即使一個(gè)處理消息的進(jìn)程掛掉，加入隊(duì)列中的消息仍然可以在系統(tǒng)恢復(fù)后被處理。
-
數(shù)據(jù)緩沖
- 有助于控制和優(yōu)化數(shù)據(jù)流經(jīng)過系統(tǒng)的速度，解決生產(chǎn)消息和消費(fèi)消息的處理速度不一致的情況。
-
靈活性 & 峰值處理能力
-
在訪問量劇增的情況下，應(yīng)用仍然需要繼續(xù)發(fā)揮作用，但是這樣的突發(fā)流量并不常見。
-
如果為以能處理這類峰值訪問為標(biāo)準(zhǔn)來投入資源隨時(shí)待命無疑是巨大的浪費(fèi)。
-
使用消息隊(duì)列能夠使關(guān)鍵組件頂住突發(fā)的訪問壓力，而不會(huì)因?yàn)橥话l(fā)的超負(fù)荷的請(qǐng)求而完全崩潰。
-
-
異步通信
-
很多時(shí)候，用戶不想也不需要立即處理消息。消息隊(duì)列提供了異步處理機(jī)制，允許用戶把一個(gè)消息放入隊(duì)列，但并不立即處理它。
-
想向隊(duì)列中放入多少消息就放多少，然后在需要的時(shí)候再去處理它們。
-
3.4、消息隊(duì)列的兩種模式
（1）點(diǎn)對(duì)點(diǎn)模式（一對(duì)一，消費(fèi)者主動(dòng)拉取數(shù)據(jù)，消息收到后消息清除）
消息生產(chǎn)者生產(chǎn)消息發(fā)送到消息隊(duì)列中，然后消息消費(fèi)者從消息隊(duì)列中取出并且消費(fèi)消息。消息被消費(fèi)以后，消息隊(duì)列中不再有存儲(chǔ)，所以消息消費(fèi)者不可能消費(fèi)到已經(jīng)被消費(fèi)的消息。消息隊(duì)列支持存在多個(gè)消費(fèi)者，但是對(duì)一個(gè)消息而言，只會(huì)有一個(gè)消費(fèi)者可以消費(fèi)。
（2）發(fā)布/訂閱模式（一對(duì)多，又叫觀察者模式，消費(fèi)者消費(fèi)數(shù)據(jù)之后不會(huì)清除消息）
消息生產(chǎn)者（發(fā)布）將消息發(fā)布到 topic 中，同時(shí)有多個(gè)消息消費(fèi)者（訂閱）消費(fèi)該消息。和點(diǎn)對(duì)點(diǎn)方式不同，發(fā)布到 topic 的消息會(huì)被所有訂閱者消費(fèi)。
發(fā)布/訂閱模式是定義對(duì)象間一種一對(duì)多的依賴關(guān)系，使得每當(dāng)一個(gè)對(duì)象（目標(biāo)對(duì)象）的狀態(tài)發(fā)生改變，則所有依賴于它的對(duì)象（觀察者對(duì)象）都會(huì)得到通知并自動(dòng)更新。
四、Kafka
4.1、Kafka 定義
Kafka 是一個(gè)分布式的基于發(fā)布/訂閱模式的消息隊(duì)列（MQ，Message Queue），主要應(yīng)用于大數(shù)據(jù)領(lǐng)域的實(shí)時(shí)計(jì)算以及日志收集。
4.2、Kafka 簡(jiǎn)介
Kafka 是最初由 Linkedin 公司開發(fā)，是一個(gè)分布式、支持分區(qū)的（partition）、多副本的（replica），基于 Zookeeper 協(xié)調(diào)的分布式消息中間件系統(tǒng)，
它的最大的特性就是可以實(shí)時(shí)的處理大量數(shù)據(jù)以滿足各種需求場(chǎng)景，比如基于 hadoop 的批處理系統(tǒng)、低延遲的實(shí)時(shí)系統(tǒng)、Spark/Flink 流式處理引擎，nginx 訪問日志，消息服務(wù)等等，用 scala 語言編寫，
Linkedin 于 2010 年貢獻(xiàn)給了 Apache 基金會(huì)并成為頂級(jí)開源項(xiàng)目。
4.3、Kafka 的特性
- 高吞吐量、低延遲
Kafka 每秒可以處理幾十萬條消息，它的延遲最低只有幾毫秒。
每個(gè) topic 可以分多個(gè) Partition，Consumer Group 對(duì) Partition 進(jìn)行消費(fèi)操作，提高負(fù)載均衡能力和消費(fèi)能力。
-
可擴(kuò)展性
kafka 集群支持熱擴(kuò)展 -
持久性、可靠性
消息被持久化到本地磁盤，并且支持?jǐn)?shù)據(jù)備份防止數(shù)據(jù)丟失 -
容錯(cuò)性
允許集群中節(jié)點(diǎn)失?。ǘ喔北厩闆r下，若副本數(shù)量為 n，則允許 n-1 個(gè)節(jié)點(diǎn)失?。?/p> -
高并發(fā)
支持?jǐn)?shù)千個(gè)客戶端同時(shí)讀寫
4.4、Kafka 系統(tǒng)架構(gòu)
-
Broker
一臺(tái) kafka 服務(wù)器就是一個(gè) broker。一個(gè)集群由多個(gè) broker 組成。一個(gè) broker 可以容納多個(gè) topic。 -
Topic
可以理解為一個(gè)隊(duì)列，生產(chǎn)者和消費(fèi)者面向的都是一個(gè) topic。
類似于數(shù)據(jù)庫(kù)的表名或者 ES 的 index
物理上不同 topic 的消息分開存儲(chǔ) -
Partition
為了實(shí)現(xiàn)擴(kuò)展性，一個(gè)非常大的 topic 可以分布到多個(gè) broker（即服務(wù)器）上，一個(gè) topic 可以分割為一個(gè)或多個(gè) partition，每個(gè) partition 是一個(gè)有序的隊(duì)列。Kafka 只保證 partition 內(nèi)的記錄是有序的，而不保證 topic 中不同 partition 的順序。
。
每個(gè) topic 至少有一個(gè) partition，當(dāng)生產(chǎn)者產(chǎn)生數(shù)據(jù)的時(shí)候，會(huì)根據(jù)分配策略選擇分區(qū)，然后將消息追加到指定的分區(qū)的隊(duì)列末尾。
4.4.1、Partation 數(shù)據(jù)路由規(guī)則：
1．指定了 patition，則直接使用；
2．未指定 patition 但指定 key（相當(dāng)于消息中某個(gè)屬性），通過對(duì) key 的 value 進(jìn)行 hash 取模，選出一個(gè) patition；
3．patition 和 key 都未指定，使用輪詢選出一個(gè) patition。
每條消息都會(huì)有一個(gè)自增的編號(hào)，用于標(biāo)識(shí)消息的偏移量，標(biāo)識(shí)順序從 0 開始。
每個(gè) partition 中的數(shù)據(jù)使用多個(gè) segment 文件存儲(chǔ)。
如果 topic 有多個(gè) partition，消費(fèi)數(shù)據(jù)時(shí)就不能保證數(shù)據(jù)的順序。嚴(yán)格保證消息的消費(fèi)順序的場(chǎng)景下（例如商品秒殺、 搶紅包），需要將 partition 數(shù)目設(shè)為 1。
4.4.2、 核心組件
-
Broker
-
Kafka服務(wù)器節(jié)點(diǎn)。
-
一個(gè)集群由多個(gè) broker 組成，一個(gè) broker 可以容納多個(gè) topic。
-
-
Consumer
-
消費(fèi)者可以從 broker 中拉取數(shù)據(jù)。
-
消費(fèi)者可以消費(fèi)多個(gè) topic 中的數(shù)據(jù)。
-
-
Consumer Group（CG）
-
消費(fèi)者組是實(shí)際的消息訂閱者。
-
消費(fèi)者組由一個(gè)或者多個(gè)消費(fèi)者，且組內(nèi)成員不能重復(fù)消費(fèi)一個(gè)Partiton數(shù)據(jù)。
-
-
Producer
- 生產(chǎn)者即數(shù)據(jù)的發(fā)布者，該角色將消息 push 發(fā)布到 Kafka 的 topic 中。
- broker 接收到生產(chǎn)者發(fā)送的消息后，broker 將該消息追加到當(dāng)前用于追加數(shù)據(jù)的 segment 文件中。
- 生產(chǎn)者發(fā)送的消息，存儲(chǔ)到一個(gè) partition 中，生產(chǎn)者也可以指定數(shù)據(jù)存儲(chǔ)的 partition。
-
Topic
可以理解為一個(gè)隊(duì)列，生產(chǎn)者和消費(fèi)者面向的都是一個(gè) topic。 -
Zookeeper
-
存儲(chǔ)kafka集群的元數(shù)據(jù)信息，生產(chǎn)者和消費(fèi)者的動(dòng)作都需要zookeeper的管理和支持。
-
比如生產(chǎn)者推送數(shù)據(jù)到kafka集群需要先通過zk尋找到kafka服務(wù)器節(jié)點(diǎn)的位置，消費(fèi)者需要從zk獲取offset記錄的上一次消費(fèi)的位置再繼續(xù)向后消費(fèi)。
-
Producer ——> Topic消息隊(duì)列 ——> 一個(gè)或者多個(gè)Partition分區(qū) ——> 一個(gè)或者多個(gè)replica副本(Leader負(fù)責(zé)數(shù)據(jù)讀寫，follower只負(fù)責(zé)同步復(fù)制leader的數(shù)據(jù))
Consumer ——> offset偏移量(用于記錄消費(fèi)者上一次消費(fèi)的位置)
4.4.3、 概念擴(kuò)展
#Partation 數(shù)據(jù)路由規(guī)則：
1．指定了 patition，則直接使用；
2．未指定 patition 但指定 key（相當(dāng)于消息中某個(gè)屬性），通過對(duì) key 的 value 進(jìn)行 hash 取模，選出一個(gè) patition；
3．patition 和 key 都未指定，使用輪詢選出一個(gè) patition。
每條消息都會(huì)有一個(gè)自增的編號(hào)，用于標(biāo)識(shí)消息的偏移量，標(biāo)識(shí)順序從 0 開始。
每個(gè) partition 中的數(shù)據(jù)使用多個(gè) segment 文件存儲(chǔ)。
如果 topic 有多個(gè) partition，消費(fèi)數(shù)據(jù)時(shí)就不能保證數(shù)據(jù)的順序。
嚴(yán)格保證消息的消費(fèi)順序的場(chǎng)景下（例如商品秒殺、 搶紅包），需要將 partition 數(shù)目設(shè)為 1。
broker 存儲(chǔ) topic 的數(shù)據(jù)。如果某 topic 有 N 個(gè) partition，集群有 N 個(gè) broker，那么每個(gè) broker 存儲(chǔ)該 topic 的一個(gè) partition。
如果某 topic 有 N 個(gè) partition，集群有 (N+M) 個(gè) broker，那么其中有 N 個(gè) broker 存儲(chǔ) topic 的一個(gè) partition， 剩下的 M 個(gè) broker 不存儲(chǔ)該 topic 的 partition 數(shù)據(jù)。
如果某 topic 有 N 個(gè) partition，集群中 broker 數(shù)目少于 N 個(gè)，那么一個(gè) broker 存儲(chǔ)該 topic 的一個(gè)或多個(gè) partition。
在實(shí)際生產(chǎn)環(huán)境中，盡量避免這種情況的發(fā)生，這種情況容易導(dǎo)致 Kafka 集群數(shù)據(jù)不均衡。
#分區(qū)的原因
方便在集群中擴(kuò)展，每個(gè)Partition可以通過調(diào)整以適應(yīng)它所在的機(jī)器，而一個(gè)topic又可以有多個(gè)Partition組成，因此整個(gè)集群就可以適應(yīng)任意大小的數(shù)據(jù)了；
可以提高并發(fā)，因?yàn)榭梢砸?span id="n5n3t3z" class="token class-name">Partition為單位讀寫了。
4.5、怎么部署kafka?
-
說明用的kafka版本，比如 我們公司用的2.6版本；
-
先部署zookeeper集群（奇數(shù)個(gè)節(jié)點(diǎn)的）
-
再在每個(gè)zk節(jié)點(diǎn)上部署kafka應(yīng)用。
-
如果是3.0版本，由于元信息儲(chǔ)存在本地，不需要zk，可以直接部署。
五、Kafka集群部署
下載Kafka軟件包
官方下載地址：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
安裝Kafka
解壓軟件包
cd /opt/
#解壓
tar zxvf kafka_2.13-2.8.2.tgz
mv kafka_2.13-2.8.2/usr/local/kafka
修改配置文件
#備份
cd /usr/local/kafka/config/
cp server.properties{,.bak}
#修改配置文件
vim server.properties
broker.id=0
#21行，broker的全局唯一編號(hào)，每個(gè)broker不能重復(fù)，因此要在其他機(jī)器上配置 broker.id=1、broker.id=2
listeners=PLAINTEXT://192.168.11.11:9092
#31行，指定監(jiān)聽的IP和端口，如果修改每個(gè)broker的IP需區(qū)分開來，也可保持默認(rèn)配置不用修改
num.network.threads=3
#42行，broker 處理網(wǎng)絡(luò)請(qǐng)求的線程數(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ǐng)求套接字的緩沖區(qū)大小
log.dirs=/usr/local/kafka/logs #60行，kafka運(yùn)行日志存放的路徑，也是數(shù)據(jù)存放的路徑
num.partitions=1 #65行，topic在當(dāng)前broker上的默認(rèn)分區(qū)個(gè)數(shù)，會(huì)被topic創(chuàng)建時(shí)的指定參數(shù)覆蓋
num.recovery.threads.per.data.dir=1 #69行，用來恢復(fù)和清理data下數(shù)據(jù)的線程數(shù)量
log.retention.hours=168
#103行，segment文件（數(shù)據(jù)文件）保留的最長(zhǎng)時(shí)間，單位為小時(shí)，默認(rèn)為7天，超時(shí)將被刪除
log.segment.bytes=1073741824 #110行，一個(gè)segment文件最大的大小，默認(rèn)為 1G，超出將新建一個(gè)新的segment文件
zookeeper.connect=192.168.11.16:2181,192.168.11.17:2181,192.168.11.18:2181
#123行，配置連接Zookeeper集群地址
#傳輸配置文件并修改
scp server.properties 192.168.11.16:/usr/local/kafka/config/
scp server.properties 192.168.11.17:/usr/local/kafka/config/
修改環(huán)境變量
vim /etc/profile
export KAFKA_HOME=/usr/local/kafka
export PATH=$PATH:$KAFKA_HOME/bin
source /etc/profile
編寫啟動(dòng)腳本，啟動(dòng)kafka
#kafka啟動(dòng)腳本
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 啟動(dòng) ------------"
${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ī)自啟
chmod +x /etc/init.d/kafka
chkconfig --add kafka
#分別啟動(dòng) Kafka
service kafka start
六、搭建Filebeat+Kafka+ELK架構(gòu)
6.1、 部署 Zookeeper+Kafka 集群
詳見上文 Zookeeper集群部署和Kafka集群部署
6.2、 部署 Filebeat
前置準(zhǔn)備
#修改主機(jī)名
hostnamectl set-hostname filebeat
#關(guān)閉防火墻和selinux
systemctl disable firewalld --now
setenforce 0
安裝filebeat
cd /opt
#解壓
tar zxvf filebeat-6.7.2-linux-x86_64.tar.gz
mv filebeat-6.7.2 -linux-x86_64 /usr/local/filebeat
修改配置文件
cd /usr/local/filebeat
vim filebeat.yml
filebeat.input:
- type: log
enabled: true
paths:
- /var/log/httpd/access_log
tags: ["access"]
- type: log
enabled: true
paths:
- /var/log/httpd/error_log
tags: ["error"]
......
#添加輸出到 Kafka 的配置
output.kafka:
enabled: true
hosts: ["192.168.11.16:9092","192.168.11.17:9092","192.168.11.18:9092"] #指定 Kafka 集群配置
topic: "httpd" #指定 Kafka 的 topic
啟動(dòng)filebeat
#啟動(dòng) filebeat
./filebeat -e -c filebeat.yml
6.3、 部署ELK
部署ELK
詳見前面博客
6.4、 在 Logstash 組件所在節(jié)點(diǎn)上新建一個(gè) Logstash 配置文件
#切換到配置文件子目錄
cd /etc/logstash/conf.d/
#新建配置文件
vim kafka.conf
input {
kafka {
bootstrap_servers => "192.168.2.100:9092,192.168.2.102:9092,192.168.2.103:9092" #kafka集群地址
topics => "httpd" #拉取的kafka的指定topic
type => "httpd_kafka" #指定 type 字段
codec => "json" #解析json格式的日志數(shù)據(jù)
auto_offset_reset => "latest" #拉取最近數(shù)據(jù)，earliest為從頭開始拉取
decorate_events => true #傳遞給elasticsearch的數(shù)據(jù)額外增加kafka的屬性數(shù)據(jù)
}
}
output {
if "access" in [tags] {
elasticsearch {
hosts => ["192.168.11.16:9200"]
index => "httpd_access-%{+YYYY.MM.dd}"
}
}
if "error" in [tags] {
elasticsearch {
hosts => ["192.168.11.17:9200"]
index => "httpd_error-%{+YYYY.MM.dd}"
}
}
stdout { codec => rubydebug }
}
#啟動(dòng) logstash
logstash -f kafka.conf
瀏覽器訪問
http://192.168.11.16:5601
登錄 Kibana
單擊“Create Index Pattern”按鈕
添加索引“httpd_access-*”
單擊 “create” 按鈕創(chuàng)建
單擊“Create Index Pattern”按鈕
添加索引“httpd_error-*”
單擊 “create” 按鈕創(chuàng)建
文章來源：http://www.zghlxwxcb.cn/news/detail-713489.html
在這里插入代碼片
文章來源地址http://www.zghlxwxcb.cn/news/detail-713489.html
到了這里，關(guān)于Zookeeper、Kafka集群與Filebeat+Kafka+ELK架構(gòu)、部署實(shí)例的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
