網站首頁編程語言正文

Zookeeper集群 + Kafka集群

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

zookeeper概述

Zookeeperl是一個開源的分布式的，為分布式框架提供協調服務的Apache項目。

Zookeeper 工作機制

zookeeper從設計模式角度來理解：是一個基于觀察者模式設計的分布式服務管理框架，它負責存儲和管理大家都關心的數據，然后接受觀察者的注冊，一旦這些數據的狀態發生變化，Zookeeper就將負責通知已經在Zookeeper上注冊的那些觀察者做出相應的反應。也就是說zookeeper=文件系統+通知機制。

Zookeeper 特點

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

Zookeeper 數據結構

ZooKeeper數據模型的結構與Linux文件系統很類似，整體上可以看作是一棵樹，每個節點稱做一個zNode。每一個zNode默認能夠存儲1MB的數據，每個zNode都可以通過其路徑唯一標識。

zookeeper應用場景

提供的服務包括：統一命名服務、統一配置管理、統一集群管理、服務器節點動態上下線、軟負載均衡等。

統一命名服務

在分布式環境下，經常需要對應用/服務進行統一命名，便于識別。例如：IP不容易記住，而域名容易記住。

統一配置管理

（1）分布式環境下，配置文件同步非常常見。一般要求一個集群中，所有節點的配置信息是一致的，比如Kafka集群。對配置文件修改后，希望能夠快速同步到（2）配置管理可交由zooKeeper實現。可將配置信息寫入ZooKeeper上的一個znode。各個客戶端服務器監聽這個znode。一旦znode中的數據被修改，zooKeeper將通知各個客戶端服務器。

統一集群管理

（1）分布式環境中，實時掌握每個節點的狀態是必要的。可根據節點實時狀態做出一些調整。
（2）zooKeeper可以實現實時監控節點狀態變化。可將節點信息寫入ZooKeeper上的一個zNode。監聽這個zNode可獲取它的實時狀態變化。

服務器動態上下線

客戶端能實時洞察到服務器上下線的變化。

zookeeper選舉機制

第一次啟動選舉機制

服務器1啟動，發起一次選舉。服務器1投自己一票。此時服務器1票數一票，不夠半數以上（3票），選舉無法完成，服務器1狀態保持為LOOKING：
服務器2啟動，再發起一次選舉。服務器1和2分別投自己一票并交換選票信息：此時服務器1發現服務器2的myid比自己目前投票推舉的（服務器1）大，更改選票為推舉服務器2。此時服務器1票數0票，服務器2票數2票，沒有半數以上結果，選舉無法完成，服務器1，2狀態保持LOOKING
服務器3啟動，發起一次選舉。此時服務器1和2都會更改選票為服務器3。此次投票結果：服務器1為0票，服務器2為0票，服務器3為3票。此時服務器3的票數已經超過半數，服務器3當選Leader。服務器1，2更改狀態為FOLLOWING，服務器3更改狀態為LEADING；
服務器4啟動，發起一次選舉。此時服務器1，2，3已經不是LOOKING狀態，不會更改選票信息。交換選票信息結果：服務器3為3票，服務器4為1票。此時服務器4服從多數，更改選票信息為服務器3，并更改狀態為FOLLOWING；
服務器5啟動，同4一樣當小弟。

非第一次啟動選舉機制

當zooKeeper 集群中的一臺服務器出現以下兩種情況之一時，就會開始進入Leader選舉：
- 服各裝初始化啟動
- 服務器運行期間無法和Leader保持連接。
而當一臺機器進入Leader選舉流程時，當前集群也可能會處于以下兩種狀態：
- 集群中本來就已經存在一個Leader
  對于已經存在Leader的情況，機器試圖去選舉Leader時，會被告知當前服務器的Leader信息，對于該機器來說，僅僅需要和Leader機器建立連接，并進行狀態同步即可。
- 集群中確實不存在Leader。
  假設zooKeeper由5臺服務器組成，SID分別為1、2、3、4、5，ZXID分別為8、8、8、7、7，并且此時sID為3的服務器是Leader。某一時刻，3和5服務器出現故障，因此開始進行Leader選舉。

選舉Leader規則

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

<span style="color:#333333"><span style="background-color:#ffffff"><code class="language-undefined">SID：服務器ID。用來唯一標識一臺ZooKeeper集群中的機器，每臺機器不能重復，和myid一致。

ZXID：事務ID。ZXID是一個事務ID，用來標識一次服務器狀態的變更。在某一時刻，集群中的每臺機器的ZXID值不一定完全一致，這和ZooKeeper服務器對于客戶端更新請求”的處理邏輯速度有關。

Epoch：每個Leader任期的代號。沒有Leader時同一輪投票過程中的邏輯時鐘值是相同的。每投完一次票這個數據就會增加
</code></span></span>

部署 Zookeeper 集群

準備 3 臺服務器做 Zookeeper 集群

192.168.239.40
192.168.239.50
192.168.239.60

關閉防火墻

<span style="color:#333333"><span style="background-color:#ffffff"><code class="language-bash"><span style="color:#008000">#所有節點執行</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">我已經安裝好了
<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>

修改配置文件（所有節點）

<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服務器與客戶端心跳時間，單位毫秒</span>
initLimit=10    <span style="color:#008000">#Leader和Follower初始連接時能容忍的最多心跳數（tickTime的數量），這里表示為10*2s</span>
syncLimit=5     <span style="color:#008000">#Leader和Follower之間同步通信的超時時間，這里表示如果超過5*2s，Leader認為Follwer死掉，并從服務器列表中刪除Follwer</span>
dataDir=/usr/local/zookeeper-3.6.3/data      ●修改，指定保存Zookeeper中的數據的目錄，目錄需要單獨創建
dataLogDir=/usr/local/zookeeper-3.6.3/logs   ●添加，指定存放日志的目錄，目錄需要單獨創建
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">#集群節點通信時使用端口3188,選舉leader時使用的端口3288</span>
-------------------------------------------------------------------------------------

server.A=B:C:D
●A是一個數字，表示這個是第幾號服務器。集群模式下需要在zoo.cfg中dataDir指定的目錄下創建一個文件myid，這個文件里面有一個數據就是A的值，Zookeeper啟動時讀取此文件，拿到里面的數據與zoo.cfg里面的配置信息比較從而判斷到底是哪個server。
●B是這個服務器的地址。
●C是這個服務器Follower與集群中的Leader服務器交換信息的端口。

●D是萬一集群中的Leader服務器掛了，需要一個端口來重新進行選舉，選出一個新的Leader，而這個端口就是用來執行選舉時服務器相互通信的端口。
</code></span></span>

拷貝配置好的 Zookeeper 配置文件到其他機器上

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

在每個節點上創建數據目錄和日志目錄

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

在每個節點的dataDir指定的目錄下創建一個 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 狀態 ------------"</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">#將服務控制腳本傳輸到其他節點</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">#設置開機自啟</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">#查看當前狀態</span>
service zookeeper status
</code></span></span>

Kafka 概述

為什么需要消息隊列（MQ）

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

當前比較常見的中間件

web應用的中間件：nginx、tomcat、apache、haproxy、squid、varnish。

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

使用消息隊列的好處

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

?

消息隊列的兩種模式

點對點模式（一對一，消費者主動拉取數據，消息收到后消息清除）
- 消息生產者生產消息發送到消息隊列中，然后消息消費者從消息隊列中取出并且消費消息。消息被消費以后，消息隊列中不再有存儲，所以消息消費者不可能消費到已經被消費的消息。消息隊列支持存在多個消費者，但是對一個消息而言，只會有一個消費者可以消費。

發布/訂閱模式（一對多，又叫觀察者模式，消費者消費數據之后不會清除消息）
- 消息生產者（發布）將消息發布到 topic中，同時有多個消息消費者（訂閱）消費該消息。和點對點方式不同，發布到topic的消息會被所有訂閱者消費。發布/訂閱模式是定義對象間一種一對多的依賴關系，使得每當一個對象（目標對象）的狀態發生改變，則所有依賴于它的對象（觀察者對象）都會得到通知并自動更新。

Kafka 定義

Kafka是一個分布式的基于發布/訂閱模式的消息隊列（MQ，Message Queue），主要應用于大數據實時處理領域。

Kafka 簡介

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

Kafka的特性

高吞吐量、低延遲

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

可擴展性

kafka集群支持熱擴展

持久性、可靠性

消息被持久化到本地磁盤，并且支持數據備份防止數據丟失

容錯性

允許集群中節點失敗（多副本情況下，若副本數量為n，則允許n-1個節點失敗）

高并發

支持數千個客戶端同時讀寫

Kafka系統架構

Broker

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

topic

可以理解為一個隊列，生產者和消費者面向的都是一個topic。
類似于數據庫的表名或者ES的index

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

partition

為了實現擴展性，一個非常大的topic可以分布到多個broker（即服務器）上，一個topic可以分割為一個或多個partition，每個partition是一個有序的隊列。Kafka 只保證partition內的記錄是有序的，而不保證topic 中不同partition的順序。

Partation 數據路由規則

指定了patition，則直接使用；
未指定patition但指定key（相當于消息中某個屬性），通過對key的value進行hash取模，選出一個patition；
patition 和key 都未指定，使用輪詢選出一個patition。
- 每條消息都會有一個自增的編號，用于標識消息的偏移量，標識順序從0開始。
- 每個partition中的數據使用多個segment文件存儲。
- 如果 topic 有多個partition，消費數據時就不能保證數據的順序。嚴格保證消息的消費順序的場景下（例如商品秒殺、搶紅包），需要將partition數目設為1。
broker存儲topic的數據。如果某topic有N個partition，集群有N個broker，那么每個broker存儲該topic的一個partition。
如果某topic有N個partition，集群有（N+M）個broker，那么其中有N個broker存儲topic的一個partition，剩下的M個broker不存儲該topic 的partition 數據。
如果某topic有N個partition，集群中broker 數目少于N個，那么一個broker 存儲該topic的一個或多個partition。在實際生產環境中，盡量避免這種情況的發生，這種情況容易導致Kafka集群數據不均衡。

分區的原因

方便在集群中擴展，每個Partition可以通過調整以適應它所在的機器，而一個topic又可以有多個Partition組成，因此整個集群就可以適應任意大小的數據了；
可以提高并發，因為可以以Partition為單位讀寫了。

Replica

副本，為保證集群中的某個節點發生故障時，該節點上的 partition 數據不丟失，且 kafka 仍然能夠繼續工作，kafka 提供了副本機制，一個 topic 的每個分區都有若干個副本，一個 leader 和若干個 follower。

Leader

每個 partition 有多個副本，其中有且僅有一個作為 Leader，Leader 是當前負責數據的讀寫的 partition。

Follower

Follower 跟隨 Leader，所有寫請求都通過 Leader 路由，數據變更會廣播給所有 Follower，Follower 與 Leader 保持數據同步。Follower 只負責備份，不負責數據的讀寫。
如果 Leader 故障，則從 Follower 中選舉出一個新的 Leader。
當 Follower 掛掉、卡住或者同步太慢，Leader 會把這個 Follower 從 ISR（Leader 維護的一個和 Leader 保持同步的 Follower 集合）列表中刪除，重新創建一個 Follower。

Producer

生產者即數據的發布者，該角色將消息 push 發布到 Kafka 的 topic 中。
broker 接收到生產者發送的消息后，broker 將該消息追加到當前用于追加數據的 segment 文件中。
生產者發送的消息，存儲到一個 partition 中，生產者也可以指定數據存儲的 partition。

Consumer

消費者可以從 broker 中 pull 拉取數據。消費者可以消費多個 topic 中的數據。

Consumer Group（CG）

消費者組，由多個 consumer 組成。
所有的消費者都屬于某個消費者組，即消費者組是邏輯上的一個訂閱者。可為每個消費者指定組名，若不指定組名則屬于默認的組。
將多個消費者集中到一起去處理某一個 Topic 的數據，可以更快的提高數據的消費能力。
消費者組內每個消費者負責消費不同分區的數據，一個分區只能由一個組內消費者消費，防止數據被重復讀取。
消費者組之間互不影響。

offset?偏移量

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

Zookeeper

Kafka 通過 Zookeeper 來存儲集群的 meta 信息。由于 consumer 在消費過程中可能會出現斷電宕機等故障，consumer 恢復后，需要從故障前的位置的繼續消費，所以 consumer 需要實時記錄自己消費到了哪個 offset，以便故障恢復后繼續消費。Kafka 0.9 版本之前，consumer 默認將 offset 保存在 Zookeeper 中；從 0.9 版本開始，consumer 默認將 offset 保存在 Kafka 一個內置的 topic 中，該 topic 為__consumer_offsets。也就是說，zookeeper的作用就是，生產者push數據到kafka集群，就必須要找到kafka集群的節點在哪里，這些都是通過zookeeper去尋找的。消費者消費哪一條數據，也需要zookeeper的支持，從zookeeper獲得offset，offset記錄上一次消費的數據消費到哪里，這樣就可以接著下一條數據進行消費。

部署 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不能重復，因此要在其他機器上配置 broker.id=1、broker.id=2</span>
listeners=PLAINTEXT://192.168.239.40:9092    
<span style="color:#008000">#31行，指定監聽的IP和端口，如果修改每個broker的IP需區分開來，也可保持默認配置不用修改</span>
num.network.threads=3    <span style="color:#008000">#42行，broker 處理網絡請求的線程數量，一般情況下不需要去修改</span>
num.io.threads=8         <span style="color:#008000">#45行，用來處理磁盤IO的線程數量，數值應該大于硬盤數</span>
socket.send.buffer.bytes=102400       <span style="color:#008000">#48行，發送套接字的緩沖區大小</span>
socket.receive.buffer.bytes=102400    <span style="color:#008000">#51行，接收套接字的緩沖區大小</span>
socket.request.max.bytes=104857600    <span style="color:#008000">#54行，請求套接字的緩沖區大小</span>
log.dirs=/usr/local/kafka/logs        <span style="color:#008000">#60行，kafka運行日志存放的路徑，也是數據存放的路徑</span>
num.partitions=1    <span style="color:#008000">#65行，topic在當前broker上的默認分區個數，會被topic創建時的指定參數覆蓋</span>
num.recovery.threads.per.data.dir=1    <span style="color:#008000">#69行，用來恢復和清理data下數據的線程數量</span>
log.retention.hours=168    <span style="color:#008000">#103行，segment文件（數據文件）保留的最長時間，單位為小時，默認為7天，超時將被刪除</span>
log.segment.bytes=1073741824    <span style="color:#008000">#110行，一個segment文件最大的大小，默認為 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傳輸到其他節點

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

修改其他節點配置文件

<span style="color:#333333"><span style="background-color:#ffffff"><code class="language-bash"><span style="color:#008000">#50節點</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行監聽地址</span>
listeners=PLAINTEXT://192.168.239.50:9092

<span style="color:#008000">#60節點</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行監聽地址</span>
listeners=PLAINTEXT://192.168.239.60:9092
</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/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 服務控制腳本

<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 狀態 ------------"</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>

將服務控制腳本傳輸到其他節點

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

設置開機自啟并啟動 Kafka（所有節點）

<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">#查看服務是否啟動</span>
</code></span></span>

Kafka 命令行操作

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

查看當前服務器中的所有 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>

發布消息

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

消費消息

<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：會把主題中以往所有的數據都讀取出來</span>
</code></span></span>

修改分區數

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

總結

kafka架構

broker：?kafka服務器一個kafka由多個broker組成

topic：?一個消息隊列生產者和消費者面向的都是topic

producer：?生產者push 推送消息數據到broker的topic中

consumer:?消費者pull從broker的topic中拉取消息數據

partition:?分區一個topic可以被分成一個或者多個partition分區，用來加快消息的傳輸（讀寫）副本對partition 進行備份Leader 負責讀寫，follower負責備例

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

offset:?偏移量記錄消費者消費消息的位置，記錄消費者上一次消費的數據到那里了，這樣可以接著下一條數據繼續進行消費

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

zookeeper 結合kafka：生產者推送數據到kafka集群時需要通過zk尋址kafka的位置，消費者消費哪條數據也需要zk支持，因為可以從zk中獲得offset

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

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

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

網站首頁編程語言正文

Zookeeper集群 + Kafka集群

zookeeper概述

Zookeeper 工作機制

Zookeeper 特點

Zookeeper 數據結構

zookeeper應用場景

zookeeper選舉機制

Kafka 概述

為什么需要消息隊列（MQ）

當前比較常見的中間件

使用消息隊列的好處

消息隊列的兩種模式

Kafka 定義

Kafka 簡介

Kafka的特性

Kafka系統架構

總結

kafka架構

相關推薦

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

網站首頁 編程語言 正文

Zookeeper集群 + Kafka集群

zookeeper概述

Zookeeper 工作機制

Zookeeper 特點

Zookeeper 數據結構

zookeeper應用場景

zookeeper選舉機制

Kafka 概述

為什么需要消息隊列（MQ）

當前比較常見的中間件

使用消息隊列的好處

消息隊列的兩種模式

Kafka 定義

Kafka 簡介

Kafka的特性

Kafka系統架構

總結

kafka架構

相關推薦

網站首頁編程語言正文