網(wǎng)站首頁編程語言正文

Zookeeper集群 + Kafka集群

作者：wmc995620 更新時間： 2022-07-13 編程語言

zookeeper概述

Zookeeperl是一個開源的分布式的，為分布式框架提供協(xié)調(diào)服務(wù)的Apache項(xiàng)目。

Zookeeper 工作機(jī)制

zookeeper從設(shè)計(jì)模式角度來理解：是一個基于觀察者模式設(shè)計(jì)的分布式服務(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ī)制。

Zookeeper 特點(diǎn)

Zookeeper：一個領(lǐng)導(dǎo)者（Leader），多個跟隨者（Fo1lower）組成的集群。
zookeepe集群中只要有半數(shù)以上節(jié)點(diǎn)存活，ookeeper集群就能正常服務(wù)。所以zookeeper適合安裝奇數(shù)臺服務(wù)器。
全局?jǐn)?shù)據(jù)一致：每個server保存一份相同的數(shù)據(jù)副本，Client無論連接到哪個server，數(shù)據(jù)都是一致的。
更新請求順序執(zhí)行，來自同一個Client的更新請求按其發(fā)送順序依次執(zhí)行，即先進(jìn)先出。
數(shù)據(jù)更新原子性，一次數(shù)據(jù)更新要么成功，要么失敗。
實(shí)時性，在一定時間范圍內(nèi)，Client能讀到最新數(shù)據(jù)。

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)識。

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集群。對配置文件修改后，希望能夠快速同步到（2）配置管理可交由zooKeeper實(shí)現(xiàn)。可將配置信息寫入ZooKeeper上的一個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ù)器上下線的變化。

zookeeper選舉機(jī)制

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

服務(wù)器1啟動，發(fā)起一次選舉。服務(wù)器1投自己一票。此時服務(wù)器1票數(shù)一票，不夠半數(shù)以上（3票），選舉無法完成，服務(wù)器1狀態(tài)保持為LOOKING：
服務(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
服務(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；
服務(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；
服務(wù)器5啟動，同4一樣當(dāng)小弟。

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

當(dāng)zooKeeper 集群中的一臺服務(wù)器出現(xiàn)以下兩種情況之一時，就會開始進(jìn)入Leader選舉：
- 服各裝初始化啟動
- 服務(wù)器運(yùn)行期間無法和Leader保持連接。
而當(dāng)一臺機(jī)器進(jìn)入Leader選舉流程時，當(dāng)前集群也可能會處于以下兩種狀態(tài)：
- 集群中本來就已經(jīng)存在一個Leader
  對于已經(jīng)存在Leader的情況，機(jī)器試圖去選舉Leader時，會被告知當(dāng)前服務(wù)器的Leader信息，對于該機(jī)器來說，僅僅需要和Leader機(jī)器建立連接，并進(jìn)行狀態(tài)同步即可。
- 集群中確實(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ī)則

EPOCH大的直接勝出
EPOCH相同，事務(wù)id大的勝出3.事務(wù)id相同，服務(wù)器id大的勝出

<span style="color:#333333"><span style="background-color:#ffffff"><code class="language-undefined">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ù)就會增加
</code></span></span>

部署 Zookeeper 集群

準(zhǔn)備 3 臺服務(wù)器做 Zookeeper 集群

192.168.239.40
192.168.239.50
192.168.239.60

關(guān)閉防火墻

<span style="color:#333333"><span style="background-color:#ffffff"><code class="language-bash"><span style="color:#008000">#所有節(jié)點(diǎn)執(zhí)行</span>
systemctl stop firewalld
systemctl <span style="color:#0000ff">disable</span> firewalld
setenforce 0
</code></span></span>

安裝 JDK

<span style="color:#333333"><span style="background-color:#ffffff"><code class="language-bash"><span style="color:#008000">#非最小化安裝一般自帶</span>
yum install -y java-1.8.0-openjdk java-1.8.0-openjdk-devel
java -version
</code></span></span>

下載安裝包

官方下載地址：Index of /dist/zookeeper

<span style="color:#333333"><span style="background-color:#ffffff"><code class="language-bash">我已經(jīng)安裝好了
<span style="color:#0000ff">cd</span> /opt
wget https://archive.apache.org/dist/zookeeper/zookeeper-3.6.3/apache-zookeeper-3.6.3-bin.tar.gz
</code></span></span>

安裝 Zookeeper

<span style="color:#333333"><span style="background-color:#ffffff"><code class="language-bash"><span style="color:#0000ff">cd</span> /opt
tar -zxvf apache-zookeeper-3.6.3-bin.tar.gz
<span style="color:#0000ff">mv</span> apache-zookeeper-3.6.3-bin /usr/local/zookeeper-3.6.3
</code></span></span>

修改配置文件（所有節(jié)點(diǎn)）

<span style="color:#333333"><span style="background-color:#ffffff"><code class="language-bash"><span style="color:#0000ff">cd</span> /usr/local/zookeeper-3.6.3/conf/
<span style="color:#0000ff">cp</span> zoo_sample.cfg zoo.cfg

vim zoo.cfg
tickTime=2000   <span style="color:#008000">#通信心跳時間，Zookeeper服務(wù)器與客戶端心跳時間，單位毫秒</span>
initLimit=10    <span style="color:#008000">#Leader和Follower初始連接時能容忍的最多心跳數(shù)（tickTime的數(shù)量），這里表示為10*2s</span>
syncLimit=5     <span style="color:#008000">#Leader和Follower之間同步通信的超時時間，這里表示如果超過5*2s，Leader認(rèn)為Follwer死掉，并從服務(wù)器列表中刪除Follwer</span>
dataDir=/usr/local/zookeeper-3.6.3/data      ●修改，指定保存Zookeeper中的數(shù)據(jù)的目錄，目錄需要單獨(dú)創(chuàng)建
dataLogDir=/usr/local/zookeeper-3.6.3/logs   ●添加，指定存放日志的目錄，目錄需要單獨(dú)創(chuàng)建
clientPort=2181   <span style="color:#008000">#客戶端連接端口</span>
<span style="color:#008000">#添加集群信息</span>
server.1=192.168.239.40:3188:3288
server.2=192.168.239.50:3188:3288
server.3=192.168.239.60:3188:3288
<span style="color:#008000">#集群節(jié)點(diǎn)通信時使用端口3188,選舉leader時使用的端口3288</span>
-------------------------------------------------------------------------------------

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ù)器相互通信的端口。
</code></span></span>

拷貝配置好的 Zookeeper 配置文件到其他機(jī)器上

<span style="color:#333333"><span style="background-color:#ffffff"><code class="language-bash">scp zoo.cfg 192.168.239.50:`<span style="color:#0000ff">pwd</span>`
scp zoo.cfg 192.168.239.60:`<span style="color:#0000ff">pwd</span>`
</code></span></span>

在每個節(jié)點(diǎn)上創(chuàng)建數(shù)據(jù)目錄和日志目錄

<span style="color:#333333"><span style="background-color:#ffffff"><code class="language-bahs">mkdir /usr/local/zookeeper-3.6.3/data
mkdir /usr/local/zookeeper-3.6.3/logs
</code></span></span>

在每個節(jié)點(diǎn)的dataDir指定的目錄下創(chuàng)建一個 myid 的文件

<span style="color:#333333"><span style="background-color:#ffffff"><code class="language-bahs">echo 1 > /usr/local/zookeeper-3.6.3/data/myid
echo 2 > /usr/local/zookeeper-3.6.3/data/myid
echo 3 > /usr/local/zookeeper-3.6.3/data/myid
</code></span></span>

配置 Zookeeper 啟動腳本

<span style="color:#333333"><span style="background-color:#ffffff"><code class="language-bash">vim /etc/init.d/zookeeper
<span style="color:#008000">#!/bin/bash</span>
<span style="color:#008000">#chkconfig:2345 20 90</span>
<span style="color:#008000">#description:Zookeeper Service Control Script</span>
ZK_HOME=<span style="color:#a31515">'/usr/local/zookeeper-3.6.3'</span>
<span style="color:#0000ff">case</span> <span style="color:#008000">$1</span> <span style="color:#0000ff">in</span>
start)
	<span style="color:#0000ff">echo</span> <span style="color:#a31515">"---------- zookeeper 啟動 ------------"</span>
	<span style="color:#008000">$ZK_HOME</span>/bin/zkServer.sh start
;;
stop)
	<span style="color:#0000ff">echo</span> <span style="color:#a31515">"---------- zookeeper 停止 ------------"</span>
	<span style="color:#008000">$ZK_HOME</span>/bin/zkServer.sh stop
;;
restart)
	<span style="color:#0000ff">echo</span> <span style="color:#a31515">"---------- zookeeper 重啟 ------------"</span>
	<span style="color:#008000">$ZK_HOME</span>/bin/zkServer.sh restart
;;
status)
	<span style="color:#0000ff">echo</span> <span style="color:#a31515">"---------- zookeeper 狀態(tài) ------------"</span>
	<span style="color:#008000">$ZK_HOME</span>/bin/zkServer.sh status
;;
*)
    <span style="color:#0000ff">echo</span> <span style="color:#a31515">"Usage: <span style="color:#008000">$0</span> {start|stop|restart|status}"</span>
<span style="color:#0000ff">esac</span>
<span style="color:#008000">#將服務(wù)控制腳本傳輸?shù)狡渌?jié)點(diǎn)</span>
scp /etc/init.d/zookeeper 192.168.239.40:/etc/init.d/
scp /etc/init.d/zookeeper 192.168.239.50:/etc/init.d/
<span style="color:#008000">#設(shè)置開機(jī)自啟</span>
<span style="color:#0000ff">chmod</span> +x /etc/init.d/zookeeper
chkconfig --add zookeeper

<span style="color:#008000">#分別啟動 Zookeeper</span>
service zookeeper start

<span style="color:#008000">#查看當(dāng)前狀態(tài)</span>
service zookeeper status
</code></span></span>

Kafka 概述

為什么需要消息隊(duì)列（MQ）

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

當(dāng)前比較常見的中間件

web應(yīng)用的中間件：nginx、tomcat、apache、haproxy、squid、varnish。

MQ消息隊(duì)列中間件：ActiveMQ、RabbitMQ、RocketMQ、Kafka、redis等。

使用消息隊(duì)列的好處

解耦
允許你獨(dú)立的擴(kuò)展或修改兩邊的處理過程，只要確保它們遵守同樣的接口約束。
可恢復(fù)性
系統(tǒng)的一部分組件失效時，不會影響到整個系統(tǒng)。消息隊(duì)列降低了進(jìn)程間的耦合度，所以即使一個處理消息的進(jìn)程掛掉，加入隊(duì)列中的消息仍然可以在系統(tǒng)恢復(fù)后被處理。
緩沖
有助于控制和優(yōu)化數(shù)據(jù)流經(jīng)過系統(tǒng)的速度，解決生產(chǎn)消息和消費(fèi)消息的處理速度不一致的情況。
靈活性&峰值處理能力
在訪問量劇增的情況下，應(yīng)用仍然需要繼續(xù)發(fā)揮作用，但是這樣的突發(fā)流量并不常見。如果為以能處理這類峰值訪問為標(biāo)準(zhǔn)來投入資源隨時待命無疑是巨大的浪費(fèi)。使用消息隊(duì)列能夠使關(guān)鍵組件頂住突發(fā)的訪問壓力，而不會因?yàn)橥话l(fā)的超負(fù)荷的請求而完全崩潰。
異步通信
很多時候，用戶不想也不需要立即處理消息。消息隊(duì)列提供了異步處理機(jī)制，允許用戶把一個消息放入隊(duì)列，但并不立即處理它。想向隊(duì)列中放入多少消息就放多少，然后在需要的時候再去處理它們。

?

消息隊(duì)列的兩種模式

點(diǎn)對點(diǎn)模式（一對一，消費(fèi)者主動拉取數(shù)據(jù)，消息收到后消息清除）
- 消息生產(chǎn)者生產(chǎn)消息發(fā)送到消息隊(duì)列中，然后消息消費(fèi)者從消息隊(duì)列中取出并且消費(fèi)消息。消息被消費(fèi)以后，消息隊(duì)列中不再有存儲，所以消息消費(fèi)者不可能消費(fèi)到已經(jīng)被消費(fèi)的消息。消息隊(duì)列支持存在多個消費(fèi)者，但是對一個消息而言，只會有一個消費(fèi)者可以消費(fèi)。

發(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ā)生改變，則所有依賴于它的對象（觀察者對象）都會得到通知并自動更新。

Kafka 定義

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

Kafka 簡介

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 基金會并成為頂級開源項(xiàng)目。

Kafka的特性

高吞吐量、低延遲

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

可擴(kuò)展性

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

持久性、可靠性

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

容錯性

允許集群中節(jié)點(diǎn)失敗（多副本情況下，若副本數(shù)量為n，則允許n-1個節(jié)點(diǎn)失敗）

高并發(fā)

支持?jǐn)?shù)千個客戶端同時讀寫

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

Broker

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

topic

可以理解為一個隊(duì)列，生產(chǎn)者和消費(fèi)者面向的都是一個topic。
類似于數(shù)據(jù)庫的表名或者ES的index

物理上不同topic的消息分開存儲

partition

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

Partation 數(shù)據(jù)路由規(guī)則

指定了patition，則直接使用；
未指定patition但指定key（相當(dāng)于消息中某個屬性），通過對key的value進(jìn)行hash取模，選出一個patition；
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ù)不均衡。

分區(qū)的原因

方便在集群中擴(kuò)展，每個Partition可以通過調(diào)整以適應(yīng)它所在的機(jī)器，而一個topic又可以有多個Partition組成，因此整個集群就可以適應(yīng)任意大小的數(shù)據(jù)了；
可以提高并發(fā)，因?yàn)榭梢砸訮artition為單位讀寫了。

Replica

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

Leader

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

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。

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。

Consumer

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

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)者組之間互不影響。

offset?偏移量

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

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)。

部署 kafka 集群

下載安裝包

官方下載地址：Apache Kafka

<span style="color:#333333"><span style="background-color:#ffffff"><code class="language-bash"><span style="color:#0000ff">cd</span> /opt
wget https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.7.1/kafka_2.13-2.7.1.tgz
</code></span></span>

安裝 Kafka

<span style="color:#333333"><span style="background-color:#ffffff"><code class="language-bash"><span style="color:#0000ff">cd</span> /opt/
tar zxvf kafka_2.13-2.7.1.tgz
<span style="color:#0000ff">mv</span> kafka_2.13-2.7.1 /usr/local/kafka
</code></span></span>

修改配置文件

<span style="color:#333333"><span style="background-color:#ffffff"><code class="language-bash"><span style="color:#0000ff">cd</span> /usr/local/kafka/config/
<span style="color:#0000ff">cp</span> server.properties{,.bak}

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

<span style="color:#0000ff">mkdir</span> /usr/local/kafka/logs
</code></span></span>

將kafka傳輸?shù)狡渌?jié)點(diǎn)

<span style="color:#333333"><span style="background-color:#ffffff"><code class="language-bash"><span style="color:#0000ff">cd</span> /usr/local
scp -r kafka/ 192.168.236.50:`<span style="color:#0000ff">pwd</span>`
scp -r kafka/ 192.168.236.60:`<span style="color:#0000ff">pwd</span>
</code></span></span>

修改其他節(jié)點(diǎn)配置文件

<span style="color:#333333"><span style="background-color:#ffffff"><code class="language-bash"><span style="color:#008000">#50節(jié)點(diǎn)</span>
<span style="color:#0000ff">cd</span> /usr/local/kafka/config/
 vim server.properties
<span style="color:#008000">#修改21行broker的全局唯一編號</span>
 broker.id=1
<span style="color:#008000">#修改31行監(jiān)聽地址</span>
listeners=PLAINTEXT://192.168.239.50:9092

<span style="color:#008000">#60節(jié)點(diǎn)</span>
<span style="color:#0000ff">cd</span> /usr/local/kafka/config/
<span style="color:#008000">#修改21行broker的全局唯一編號</span>
 broker.id=2
<span style="color:#008000">#修改31行監(jiān)聽地址</span>
listeners=PLAINTEXT://192.168.239.60:9092
</code></span></span>

修改環(huán)境變量（所有節(jié)點(diǎn)）

<span style="color:#333333"><span style="background-color:#ffffff"><code class="language-bash"><span style="color:#0000ff">cd</span> /usr/local/kafka/bin
<span style="color:#0000ff">ls</span>
vim /etc/profile
<span style="color:#0000ff">export</span> KAFKA_HOME=/usr/local/kafka
<span style="color:#0000ff">export</span> PATH=<span style="color:#008000">$PATH</span>:<span style="color:#008000">$KAFKA_HOME</span>/bin
<span style="color:#0000ff">source</span> /etc/profile
</code></span></span>

編寫 Zookeeper 服務(wù)控制腳本

<span style="color:#333333"><span style="background-color:#ffffff"><code class="language-bash">vim /etc/init.d/kafka
<span style="color:#008000">#!/bin/bash</span>
<span style="color:#008000">#chkconfig:2345 22 88</span>
<span style="color:#008000">#description:Kafka Service Control Script</span>
KAFKA_HOME=<span style="color:#a31515">'/usr/local/kafka'</span>
<span style="color:#0000ff">case</span> <span style="color:#008000">$1</span> <span style="color:#0000ff">in</span>
start)
	<span style="color:#0000ff">echo</span> <span style="color:#a31515">"---------- Kafka 啟動 ------------"</span>
	<span style="color:#008000">${KAFKA_HOME}</span>/bin/kafka-server-start.sh -daemon <span style="color:#008000">${KAFKA_HOME}</span>/config/server.properties
;;
stop)
	<span style="color:#0000ff">echo</span> <span style="color:#a31515">"---------- Kafka 停止 ------------"</span>
	<span style="color:#008000">${KAFKA_HOME}</span>/bin/kafka-server-stop.sh
;;
restart)
	<span style="color:#008000">$0</span> stop
	<span style="color:#008000">$0</span> start
;;
status)
	<span style="color:#0000ff">echo</span> <span style="color:#a31515">"---------- Kafka 狀態(tài) ------------"</span>
	count=$(ps -ef | grep kafka | egrep -cv <span style="color:#a31515">"grep|$$"</span>)
	<span style="color:#0000ff">if</span> [ <span style="color:#a31515">"<span style="color:#008000">$count</span>"</span> -eq 0 ];<span style="color:#0000ff">then</span>
        <span style="color:#0000ff">echo</span> <span style="color:#a31515">"kafka is not running"</span>
    <span style="color:#0000ff">else</span>
        <span style="color:#0000ff">echo</span> <span style="color:#a31515">"kafka is running"</span>
    <span style="color:#0000ff">fi</span>
;;
*)
    <span style="color:#0000ff">echo</span> <span style="color:#a31515">"Usage: <span style="color:#008000">$0</span> {start|stop|restart|status}"</span>
<span style="color:#0000ff">esac</span>
</code></span></span>

將服務(wù)控制腳本傳輸?shù)狡渌?jié)點(diǎn)

<span style="color:#333333"><span style="background-color:#ffffff"><code class="language-bash">scp /etc/init.d/kafka 192.168.239.50:/etc/init.d/
scp /etc/init.d/kafka 192.168.239.60:/etc/init.d/
</code></span></span>

設(shè)置開機(jī)自啟并啟動 Kafka（所有節(jié)點(diǎn)）

<span style="color:#333333"><span style="background-color:#ffffff"><code class="language-bash"><span style="color:#0000ff">chmod</span> +x /etc/init.d/kafka
chkconfig --add kafka
service kafka start
ps -ef | grep kafka  <span style="color:#008000">#查看服務(wù)是否啟動</span>
</code></span></span>

Kafka 命令行操作

創(chuàng)建topic

<span style="color:#333333"><span style="background-color:#ffffff"><code class="language-bash">kafka-topics.sh --create --zookeeper 192.168.239.40:2181,192.168.239.50:2181,192.168.239.60:2181 --replication-factor 2 --partitions 3 --topic ky18

<span style="color:#008000">#--zookeeper：定義 zookeeper 集群服務(wù)器地址，如果有多個 IP 地址使用逗號分割，一般使用一個 IP 即可</span>
<span style="color:#008000">#--replication-factor：定義分區(qū)副本數(shù)，1 代表單副本，建議為 2 </span>
<span style="color:#008000">#--partitions：定義分區(qū)數(shù) </span>
<span style="color:#008000">#--topic：定義 topic 名稱</span>
</code></span></span>

查看當(dāng)前服務(wù)器中的所有 topic

<span style="color:#333333"><span style="background-color:#ffffff"><code class="language-bash">kafka-topics.sh --list --zookeeper 192.168.239.40:2181,192.168.239.50:2181,192.168.239.60:2181 
</code></span></span>

查看某個 topic 的詳情

<span style="color:#333333"><span style="background-color:#ffffff"><code class="language-bash">kafka-topics.sh  --describe --zookeeper 192.168.239.40:2181,192.168.239.50:2181,192.168.239.60:2181 
</code></span></span>

發(fā)布消息

<span style="color:#333333"><span style="background-color:#ffffff"><code class="language-bash">kafka-console-producer.sh --broker-list 192.168.239.40:9092,192.168.239.50:9092,192.168.239.60:9092  --topic ky18
</code></span></span>

消費(fèi)消息

<span style="color:#333333"><span style="background-color:#ffffff"><code class="language-bash">kafka-console-consumer.sh --bootstrap-server 192.168.239.40:9092,192.168.239.50:9092,192.168.239.60:9092 --topic ky18 --from-beginning
<span style="color:#008000">#--from-beginning：會把主題中以往所有的數(shù)據(jù)都讀取出來</span>
</code></span></span>

修改分區(qū)數(shù)

<span style="color:#333333"><span style="background-color:#ffffff"><code class="language-bash">kafka-topics.sh --zookeeper 192.168.239.40:2181,192.168.239.50:2181,192.168.239.60:2181 --alter --topic ky18 --partitions 6
</code></span></span>

刪除 topic

<span style="color:#333333"><span style="background-color:#ffffff"><code class="language-bash">kafka-topics.sh --delete --zookeeper 192.168.239.40:2181,192.168.239.50:2181,192.168.239.60:2181 --topic ky18
</code></span></span>

總結(jié)

kafka架構(gòu)

broker：?kafka服務(wù)器一個kafka由多個broker組成

topic：?一個消息隊(duì)列生產(chǎn)者和消費(fèi)者面向的都是topic

producer：?生產(chǎn)者push 推送消息數(shù)據(jù)到broker的topic中

consumer:?消費(fèi)者pull從broker的topic中拉取消息數(shù)據(jù)

partition:?分區(qū)一個topic可以被分成一個或者多個partition分區(qū)，用來加快消息的傳輸（讀寫）副本對partition 進(jìn)行備份Leader 負(fù)責(zé)讀寫，follower負(fù)責(zé)備例

partition中的消息數(shù)據(jù)是有序的。partition之間是無序的在秒殺紅包等要求有序場景中只能使用一個partition

offset:?偏移量記錄消費(fèi)者消費(fèi)消息的位置，記錄消費(fèi)者上一次消費(fèi)的數(shù)據(jù)到那里了，這樣可以接著下一條數(shù)據(jù)繼續(xù)進(jìn)行消費(fèi)

zookeeper:?保存kafka集群的源信息，保存offset

zookeeper 結(jié)合kafka：生產(chǎn)者推送數(shù)據(jù)到kafka集群時需要通過zk尋址kafka的位置，消費(fèi)者消費(fèi)哪條數(shù)據(jù)也需要zk支持，因?yàn)榭梢詮膠k中獲得offset

原文鏈接：https://blog.csdn.net/wmc995620/article/details/125591090

上一篇：docker基本概念及安裝
下一篇：Sybase使用Spring的只讀事物報(bào) Use ‘set

日本免费高清视频-国产福利视频导航-黄色在线播放国产-天天操天天操天天操天天操|www.shdianci.com

網(wǎng)站首頁編程語言正文

Zookeeper集群 + Kafka集群

zookeeper概述

Zookeeper 工作機(jī)制

Zookeeper 特點(diǎn)

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

zookeeper應(yīng)用場景

zookeeper選舉機(jī)制

Kafka 概述

為什么需要消息隊(duì)列（MQ）

當(dāng)前比較常見的中間件

使用消息隊(duì)列的好處

消息隊(duì)列的兩種模式

Kafka 定義

Kafka 簡介

Kafka的特性

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

總結(jié)

kafka架構(gòu)

相關(guān)推薦

日本免费高清视频-国产福利视频导航-黄色在线播放国产-天天操天天操天天操天天操|www.shdianci.com

網(wǎng)站首頁 編程語言 正文

Zookeeper集群 + Kafka集群

zookeeper概述

Zookeeper 工作機(jī)制

Zookeeper 特點(diǎn)

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

zookeeper應(yīng)用場景

zookeeper選舉機(jī)制

Kafka 概述

為什么需要消息隊(duì)列（MQ）

當(dāng)前比較常見的中間件

使用消息隊(duì)列的好處

消息隊(duì)列的兩種模式

Kafka 定義

Kafka 簡介

Kafka的特性

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

總結(jié)

kafka架構(gòu)

相關(guān)推薦

網(wǎng)站首頁編程語言正文