主頁 >  其他 > 基于zookeeper的kafka中間件

基于zookeeper的kafka中間件

2022-12-14 06:56:57 其他

一、Zookeeper 概述

1.1Zookeeper 定義

Zookeeper是一個開源的分布式的，為分布式框架提供協調服務的Apache專案，

1.2Zookeeper 作業機制

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

1.3Zookeeper 特點

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

1.4Zookeeper 資料結構

ZooKeeper資料模型的結構與Linux檔案系統很類似，整體上可以看作是一棵樹，每個節點稱做一個ZNode，每一個ZNode默認能夠存盤1MB的資料，每個ZNode都可以通過其路徑唯一標識，

1.5Zookeeper 應用場景

提供的服務包括：統一命名服務、統一配置管理、統一集群管理、服務器節點動態上下線、軟負載均衡等，
●統一命名服務
在分布式環境下，經常需要對應用/服務進行統一命名，便于識別，例如：IP不容易記住，而域名容易記住，
●統一配置管理
（1）分布式環境下，組態檔同步非常常見，一般要求一個集群中，所有節點的配置資訊是一致的，比如Kafka集群，對組態檔修改后，希望能夠快速同步到各個節點上，
（2）配置管理可交由ZooKeeper實作，可將配置資訊寫入ZooKeeper上的一個Znode，各個客戶端服務器監聽這個Znode，一旦 Znode中的資料被修改，ZooKeeper將通知各個客戶端服務器，
●統一集群管理
（1）分布式環境中，實時掌握每個節點的狀態是必要的，可根據節點實時狀態做出一些調整，
（2）ZooKeeper可以實作實時監控節點狀態變化，可將節點資訊寫入ZooKeeper上的一個ZNode，監聽這個ZNode可獲取它的實時狀態變化，
●服務器動態上下線
客戶端能實時洞察到服務器上下線的變化，
●軟負載均衡
在Zookeeper中記錄每臺服務器的訪問數，讓訪問數最少的服務器去處理最新的客戶端請求，

1.6Zookeeper 選舉機制

（1）第一次啟動選舉機制

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

（2）非第一次啟動選舉機制

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

選舉Leader規則：
1.EPOCH大的直接勝出
2.EPOCH相同，事務id大的勝出
3.事務id相同，服務器id大的勝出
注釋：
SID：服務器ID，用來唯一標識一臺ZooKeeper集群中的機器，每臺機器不能重復，和myid一致，
ZXID：事務ID，ZXID是一個事務ID，用來標識一次服務器狀態的變更，在某一時刻，集群中的每臺機器的ZXID值不一定完全一致，這和ZooKeeper服務器對于客戶端“更新請求”的處理邏輯速度有關，
Epoch：每個Leader任期的代號，沒有Leader時同一輪投票程序中的邏輯時鐘值是相同的，每投完一次票這個資料就會增加

二、部署 Zookeeper 集群

2.1部署Zookeeper集群的操作步驟

準備 3 臺服務器做 Zookeeper 集群
192.168.10.10
192.168.10.20
192.168.10.30

2.1.1 安裝前準備

@關閉防火墻

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

2.1.2安裝 Zookeeper

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　　

2.1.3修改組態檔

cd /usr/local/zookeeper-3.5.7/conf/
cp zoo_sample.cfg zoo.cfg
 
vim zoo.cfg
tickTime=2000 #通信心跳時間，Zookeeper服務器與客戶端心跳時間，單位毫秒
initLimit=10 #Leader和Follower初始連接時能容忍的最多心跳數（tickTime的數量），這里表示為10*2s
syncLimit=5 #Leader和Follower之間同步通信的超時時間，這里表示如果超過5*2s，Leader認為Follwer死掉，并從服務器串列中洗掉Follwer
dataDir=/usr/local/zookeeper-3.5.7/data ●修改，指定保存Zookeeper中的資料的目錄，目錄需要單獨創建
dataLogDir=/usr/local/zookeeper-3.5.7/logs ●添加，指定存放日志的目錄，目錄需要單獨創建
clientPort=2181 #客戶端連接埠
#添加集群資訊
server.1=192.168.10.10:3188:3288
server.2=192.168.10.20:3188:3288
server.3=192.168.10.30:3188:3288　

server.A=B:C:D

  • A是一個數字，表示這個是第幾號服務器，集群模式下需要在zoo.cfg中dataDir指定的目錄下創建一個檔案myid，這個檔案里面有一個資料就是A的值，Zookeeper啟動時讀取此檔案，拿到里面的資料與zoo.cfg里面的配置資訊比較從而判斷到底是哪個server，
  • B是這個服務器的地址，
  • C是這個服務器Follower與集群中的Leader服務器交換資訊的埠，
  • D是萬一集群中的Leader服務器掛了，需要一個埠來重新進行選舉，選出一個新的Leader，而這個埠就是用來執行選舉時服務器相互通信的埠，

2.1.4拷貝配置好的 Zookeeper 組態檔到其他機器上

scp /usr/local/zookeeper-3.5.7/conf/zoo.cfg 192.168.10.20:/usr/local/zookeeper-3.5.7/conf/
scp /usr/local/zookeeper-3.5.7/conf/zoo.cfg 192.168.20.30:/usr/local/zookeeper-3.5.7/conf/

2.1.5在每個節點上創建資料目錄和日志目錄

mkdir /usr/local/zookeeper-3.5.7/data
mkdir /usr/local/zookeeper-3.5.7/logs　　

2.1.6在每個節點的dataDir指定的目錄下創建一個 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　

2.1.7配置 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 狀態 ------------"
$ZK_HOME/bin/zkServer.sh status
;;
*)
echo "Usage: $0 {start|stop|restart|status}"
esac　　　

2.1.8設定開機自啟

chmod +x /etc/init.d/zookeeper
chkconfig --add zookeeper

2.1.9分別啟動 Zookeeper

service zookeeper start　　

2.1.10查看當前狀態

service zookeeper status

2.2實體操作：部署Zookeeper集群(三臺機器)

2.2.1安裝前準備

2.2.2安裝 Zookeeper

2.2.3修改組態檔

2.2.4拷貝配置好的 Zookeeper 組態檔到其他機器上

2.2.5在每個節點上創建資料目錄和日志目錄

2.2.6在每個節點的dataDir指定的目錄下創建一個 myid 的檔案

2.2.7配置 Zookeeper 啟動腳本

2.2.8設定開機自啟并開啟服務

2.2.9查看當前狀態

一個leader，兩個follower

三、Kafka 概述

3.1為什么需要訊息佇列（MQ）

主要原因是由于在高并發環境下，同步請求來不及處理，請求往往會發生阻塞，比如大量的請求并發訪問資料庫，導致行鎖表鎖，最后請求執行緒會堆積過多，從而觸發 too many connection 錯誤，引發雪崩效應，
我們使用訊息佇列，通過異步處理請求，從而緩解系統的壓力，訊息佇列常應用于異步處理，流量削峰，應用解耦，訊息通訊等場景，
當前比較常見的 MQ 中間件有 ActiveMQ、RabbitMQ、RocketMQ、Kafka 等，

3.2使用訊息佇列的好處

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

3.3訊息佇列的兩種模式

（1）點對點模式（一對一，消費者主動拉取資料，訊息收到后訊息清除）
訊息生產者生產訊息發送到訊息佇列中，然后訊息消費者從訊息佇列中取出并且消費訊息，訊息被消費以后，訊息佇列中不再有存盤，所以訊息消費者不可能消費到已經被消費的訊息，訊息佇列支持存在多個消費者，但是對一個訊息而言，只會有一個消費者可以消費，
（2）發布/訂閱模式（一對多，又叫觀察者模式，消費者消費資料之后不會清除訊息）
訊息生產者（發布）將訊息發布到 topic 中，同時有多個訊息消費者（訂閱）消費該訊息，和點對點方式不同，發布到 topic 的訊息會被所有訂閱者消費，
發布/訂閱模式是定義物件間一種一對多的依賴關系，使得每當一個物件（目標物件）的狀態發生改變，則所有依賴于它的物件（觀察者物件）都會得到通知并自動更新，

3.4Kafka 定義

Kafka 是一個分布式的基于發布/訂閱模式的訊息佇列（MQ，Message Queue），主要應用于大資料實時處理領域，

3.5Kafka 簡介

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

3.6Kafka 的特性

（1）高吞吐量、低延遲
Kafka 每秒可以處理幾十萬條訊息，它的延遲最低只有幾毫秒，每個 topic 可以分多個 Partition，Consumer Group 對 Partition 進行消費操作，提高負載均衡能力和消費能力，
（2）可擴展性
kafka 集群支持熱擴展
（3）持久性、可靠性
訊息被持久化到本地磁盤，并且支持資料備份防止資料丟失
（4）容錯性
允許集群中節點失敗（多副本情況下，若副本數量為 n，則允許 n-1 個節點失敗）
（5）高并發
支持數千個客戶端同時讀寫

3.7Kafka 系統架構

（1）Broker：一臺 kafka 服務器就是一個 broker，一個集群由多個 broker 組成，一個 broker 可以容納多個 topic，
（2）Topic：可以理解為一個佇列，生產者和消費者面向的都是一個 topic，類似于資料庫的表名或者 ES 的 index，物理上不同 topic 的訊息分開存盤
（3）Partition：為了實作擴展性，一個非常大的 topic 可以分布到多個 broker（即服務器）上，一個 topic 可以分割為一個或多個 partition，每個 partition 是一個有序的佇列，Kafka 只保證 partition 內的記錄是有序的，而不保證 topic 中不同 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為單位讀寫了，

（4）Replica：副本，為保證集群中的某個節點發生故障時，該節點上的 partition 資料不丟失，且 kafka 仍然能夠繼續作業，kafka 提供了副本機制，一個 topic 的每個磁區都有若干個副本，一個 leader 和若干個 follower，
（5）Leader：每個 partition 有多個副本，其中有且僅有一個作為 Leader，Leader 是當前負責資料的讀寫的 partition，
（6）Follower：Follower 跟隨 Leader，所有寫請求都通過 Leader 路由，資料變更會廣播給所有 Follower，Follower 與 Leader 保持資料同步，Follower 只負責備份，不負責資料的讀寫，
如果 Leader 故障，則從 Follower 中選舉出一個新的 Leader，
當 Follower 掛掉、卡住或者同步太慢，Leader 會把這個 Follower 從 ISR（Leader 維護的一個和 Leader 保持同步的 Follower 集合） 串列中洗掉，重新創建一個 Follower，
（7）Producer：生產者即資料的發布者，該角色將訊息發布到 Kafka 的 topic 中，
broker 接收到生產者發送的訊息后，broker 將該訊息追加到當前用于追加資料的 segment 檔案中，
生產者發送的訊息，存盤到一個 partition 中，生產者也可以指定資料存盤的 partition，
（8）Consumer：消費者可以從 broker 中讀取資料，消費者可以消費多個 topic 中的資料，
（9）Consumer Group（CG）

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

（10）offset 偏移量

  • 可以唯一的標識一條訊息，
  • 偏移量決定讀取資料的位置，不會有執行緒安全的問題，消費者通過偏移量來決定下次讀取的訊息（即消費位置），
  • 訊息被消費之后，并不被馬上洗掉，這樣多個業務就可以重復使用 Kafka 的訊息，
  • 某一個業務也可以通過修改偏移量達到重新讀取訊息的目的，偏移量由用戶控制，
  • 訊息最侄訓是會被洗掉的，默認生命周期為 1 周（7*24小時），

（11）Zookeeper

  • Kafka 通過 Zookeeper 來存盤集群的 meta 資訊，
  • 由于 consumer 在消費程序中可能會出現斷電宕機等故障，consumer 恢復后，需要從故障前的位置的繼續消費，所以 consumer 需要實時記錄自己消費到了哪個 offset，以便故障恢復后繼續消費，
  • Kafka 0.9 版本之前，consumer 默認將 offset 保存在 Zookeeper 中；從 0.9 版本開始，consumer 默認將 offset 保存在 Kafka 一個內置的 topic 中，該 topic 為 __consumer_offsets，

四、部署zookeeper + kafka 集群

4.1部署zookeeper + kafka 集群

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

4.1.2安裝 Kafka

cd /opt/
tar zxvf kafka_2.13-2.7.1.tgz
mv kafka_2.13-2.7.1 /usr/local/kafka　

4.1.3修改組態檔

cd /usr/local/kafka/config/
cp server.properties{,.bak}
 
vim server.properties
broker.id=0                                    #21行，broker的全域唯一編號，每個broker不能重復，因此要在其他機器上配置 broker.id=1、broker.id=2
listeners=PLAINTEXT://192.168.10.10:9092       #31行，指定監聽的IP和埠，如果修改每個broker的IP需區分開來，也可保持默認配置不用修改
num.network.threads=3                          #42行，broker 處理網路請求的執行緒數量，一般情況下不需要去修改
num.io.threads=8                               #45行，用來處理磁盤IO的執行緒數量，數值應該大于硬碟數
socket.send.buffer.bytes=102400                #48行，發送套接字的緩沖區大小
socket.receive.buffer.bytes=102400             #51行，接收套接字的緩沖區大小
socket.request.max.bytes=104857600             #54行，請求套接字的緩沖區大小
log.dirs=/usr/local/kafka/logs                 #60行，kafka運行日志存放的路徑，也是資料存放的路徑
num.partitions=1                               #65行，topic在當前broker上的默認磁區個數，會被topic創建時的指定引數覆寫
num.recovery.threads.per.data.dir=1            #69行，用來恢復和清理data下資料的執行緒數量
log.retention.hours=168                        #103行，segment檔案（資料檔案）保留的最長時間，單位為小時，默認為7天，超時將被洗掉
log.segment.bytes=1073741824                   #110行，一個segment檔案最大的大小，默認為 1G，超出將新建一個新的segment檔案
zookeeper.connect=192.168.10.10:2181,192.168.10.20:2181,192.168.10.30:2181                                   #123行，配置連接Zookeeper集群地址

4.1.4修改環境變數

vim /etc/profile
export KAFKA_HOME=/usr/local/kafka
export PATH=$PATH:$KAFKA_HOME/bin
 
source /etc/profile　

4.1.5配置 Zookeeper 啟動腳本

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

4.1.6設定開機自啟

chmod +x /etc/init.d/kafka
chkconfig --add kafka　

4.1.7分別啟動 Kafka

service kafka start　　

4.1.8Kafka 命令列操作

@創建topic

kafka-topics.sh --create --zookeeper 192.168.10.10:2181,192.168.10.20:2181,192.168.10.30:2181 --replication-factor 2 --partitions 3 --topic test
 

--zookeeper：定義 zookeeper 集群服務器地址，如果有多個 IP 地址使用逗號分割，一般使用一個 IP 即可
--replication-factor：定義磁區副本數，1 代表單副本，建議為 2
--partitions：定義磁區數
--topic：定義 topic 名稱

@查看當前服務器中的所有 topic

kafka-topics.sh --list --zookeeper 192.168.10.10:2181,192.168.10.20:2181,192.168.10.30:2181　

@查看某個 topic 的詳情

kafka-topics.sh --describe --zookeeper 192.168.10.10:2181,192.168.10.20:2181,192.168.10.30:2181　　

@發布訊息

kafka-console-producer.sh --broker-list 192.168.10.10:9092,192.168.10.20:9092,192.168.10.30:9092 --topic test　

@消費訊息

kafka-console-consumer.sh --bootstrap-server 192.168.10.10:9092,192.168.10.20:9092,192.168.10.30:9092 --topic test --from-beginning
 
--from-beginning：會把主題中以往所有的資料都讀取出來

@修改磁區數

kafka-topics.sh --zookeeper 192.168.10.10:2181,192.168.10.20:2181,192.168.10.30:2181 --alter --topic test  --partitions 6　　

@洗掉 topic

kafka-topics.sh --delete --zookeeper 192.168.10.10:2181,192.168.10.20:2181,192.168.10.30:2181 --topic  test

4.2實體操作：部署zookeeper + kafka 集群

4.2.1安裝zookeeper集群

詳見本篇博客上文，接著上面的實驗繼續做（在所有集群服務器操作）

4.2.2下載安裝包并安裝kafka

4.2.3修改組態檔

4.2.4修改環境變數

4.2.5配置 kafka 啟動腳本并設定開機自啟，啟動kafka

4.2.6Kafka 命令列操作

@創建topic并查看

@發布訊息并讀取訊息

@修改磁區數

@洗掉 topic

五、部署 Filebeat+Kafka+ELK

5.1部署 Filebeat+Kafka+ELK的操作步驟

5.1.1部署 Zookeeper+Kafka 集群

見上文，接著上面的實驗做的

5.1.2部署 Filebeat

要搭建ELK，詳見之前的博客

安裝httpd：yum -y install httpd
　　　　　  systemctl start httpd
cd /usr/local/filebeat

vim filebeat.yml
filebeat.prospectors:
- 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.10.10:9092","192.168.10.20:9092","192.168.10.30:9092"]    #指定 Kafka 集群配置
  topic: "httpd"    #指定 Kafka 的 topic
  
#啟動 filebeat
./filebeat -e -c filebeat.yml

5.1.3部署 ELK，在 Logstash 組件所在節點上新建一個 Logstash 組態檔

vim kafka.conf
input {
    kafka {
        bootstrap_servers => "192.168.10.10:9092,192.168.10.20:9092,192.168.10.30:9092"  #kafka集群地址
        topics  => "httpd"     #拉取的kafka的指定topic
        type => "httpd_kafka"  #指定 type 欄位
        codec => "json"        #決議json格式的日志資料
        auto_offset_reset => "latest"  #拉取最近資料，earliest為從頭開始拉取
        decorate_events => true   #傳遞給elasticsearch的資料額外增加kafka的屬性資料
    }
}

output {
  if "access" in [tags] {
    elasticsearch {
      hosts => ["192.168.10.40:9200"]
      index => "httpd_access-%{+YYYY.MM.dd}"
    }
  }
  
  if "error" in [tags] {
    elasticsearch {
      hosts => ["192.168.10.40:9200"]
      index => "httpd_error-%{+YYYY.MM.dd}"
    }
  }
  
  stdout { codec => rubydebug }
}

#啟動 logstash
logstash -f kafka.conf

5.1.4啟動 logstash

logstash -f filebeat.conf　　

5.1.5瀏覽器訪問測驗

瀏覽器訪問 http://192.168.10.40:5601 登錄 Kibana，單擊“Create Index Pattern”按鈕添加索引“httpd_access-*”，單擊 “create” 按鈕創建，單擊 “Discover” 按鈕可查看圖表資訊及日志資訊，

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/539860.html

標籤：其他

上一篇：Mybatis原始碼決議之執行SQL陳述句

下一篇：性能達1.5+倍！昇騰AI助力分子動力學模擬研究

標籤雲
其他(157675) Python(38076) JavaScript(25376) Java(17977) C(15215) 區塊鏈(8255) C＃(7972) AI(7469) 爪哇(7425) MySQL(7132) html(6777) 基礎類(6313) sql(6102) 熊猫(6058) PHP(5869) 数组(5741) R(5409) Linux(5327) 反应(5209) 腳本語言(PerlPython)(5129) 非技術區(4971) Android(4554) 数据框(4311) css(4259) 节点.js(4032) C語言(3288) json(3245) 列表(3129) 扑(3119) C++語言(3117) 安卓(2998) 打字稿(2995) VBA(2789) Java相關(2746) 疑難問題(2699) 细绳(2522) 單片機工控(2479) iOS(2429) ASP.NET(2402) MongoDB(2323) 麻木的(2285) 正则表达式(2254) 字典(2211) 循环(2198) 迅速(2185) 擅长(2169) 镖(2155) 功能(1967) .NET技术(1958) Web開發(1951) python-3.x(1918) HtmlCss(1915) 弹簧靴(1913) C++(1909) xml(1889) PostgreSQL(1872) .NETCore(1853) 谷歌表格(1846) Unity3D(1843) for循环(1842)

熱門瀏覽
  • 網閘典型架構簡述

    網閘架構一般分為兩種：三主機的三系統架構網閘和雙主機的2+1架構網閘。 三主機架構分別為內端機、外端機和仲裁機。三機無論從軟體和硬體上均各自獨立。首先從硬體上來看，三機都用各自獨立的主板、記憶體及存盤設備。從軟體上來看，三機有各自獨立的作業系統。這樣能達到完全的三機獨立。對于“2+1”系統，“2”分為 ......

    uj5u.com 2020-09-10 02:00:44 more
  • 如何從xshell上傳檔案到centos linux虛擬機里

    如何從xshell上傳檔案到centos linux虛擬機里及：虛擬機CentOs下執行 yum -y install lrzsz命令，出現錯誤：鏡像無法找到軟體包 前言 一、安裝lrzsz步驟 二、上傳檔案 三、遇到的問題及解決方案 總結 前言 提示：其實很簡單，往虛擬機上安裝一個上傳檔案的工具 ......

    uj5u.com 2020-09-10 02:00:47 more
  • 一、SQLMAP入門

    一、SQLMAP入門 1、判斷是否存在注入 sqlmap.py -u 網址/id=1 id=1不可缺少。當注入點后面的引數大于兩個時。需要加雙引號， sqlmap.py -u "網址/id=1&uid=1" 2、判斷文本中的請求是否存在注入 從文本中加載http請求，SQLMAP可以從一個文本檔案中 ......

    uj5u.com 2020-09-10 02:00:50 more
  • Metasploit 簡單使用教程

    metasploit 簡單使用教程 浩先生， 2020-08-28 16:18:25 分類專欄： kail 網路安全 linux 文章標簽： linux資訊安全 編輯 著作權 metasploit 使用教程 前言 一、Metasploit是什么？ 二、準備作業 三、具體步驟 前言 Msfconsole ......

    uj5u.com 2020-09-10 02:00:53 more
  • 游戲逆向之驅動層與用戶層通訊

    驅動層代碼： #pragma once #include <ntifs.h> #define add_code CTL_CODE(FILE_DEVICE_UNKNOWN,0x800,METHOD_BUFFERED,FILE_ANY_ACCESS) /* 更多游戲逆向視頻www.yxfzedu.com ......

    uj5u.com 2020-09-10 02:00:56 more
  • 北斗電力時鐘（北斗授時服務器）讓網路資料更精準

    北斗電力時鐘（北斗授時服務器）讓網路資料更精準 北斗電力時鐘（北斗授時服務器）讓網路資料更精準 京準電子科技官微——ahjzsz 近幾年，資訊技術的得了快速發展，互聯網在逐漸普及，其在人們生活和生產中都得到了廣泛應用，并且取得了不錯的應用效果。計算機網路資訊在電力系統中的應用，一方面使電力系統的運行 ......

    uj5u.com 2020-09-10 02:01:03 more
  • 【CTF】CTFHub 技能樹 彩蛋 writeup

    ?碎碎念 CTFHub：https://www.ctfhub.com/ 筆者入門CTF時時剛開始刷的是bugku的舊平臺，后來才有了CTFHub。 感覺不論是網頁UI設計，還是題目質量，賽事跟蹤，工具軟體都做得很不錯。 而且因為獨到的金幣制度的確讓人有一種想去刷題賺金幣的感覺。 個人還是非常喜歡這個 ......

    uj5u.com 2020-09-10 02:04:05 more
  • 02windows基礎操作

    我學到了一下幾點 Windows系統目錄結構與滲透的作用 常見Windows的服務詳解 Windows埠詳解 常用的Windows注冊表詳解 hacker DOS命令詳解（net user / type /md /rd/ dir /cd /net use copy、批處理 等） 利用dos命令制作 ......

    uj5u.com 2020-09-10 02:04:18 more
  • 03.Linux基礎操作

    我學到了以下幾點 01Linux系統介紹02系統安裝，密碼啊破解03Linux常用命令04LAMP 01LINUX windows： win03 8 12 16 19 配置不繁瑣 Linux：redhat,centos(紅帽社區版)，Ubuntu server,suse unix:金融機構，證券，銀 ......

    uj5u.com 2020-09-10 02:04:30 more
  • 05HTML

    01HTML介紹 02頭部標簽講解03基礎標簽講解04表單標簽講解 HTML前段語言 js1.了解代碼2.根據代碼 懂得挖掘漏洞 （POST注入/XSS漏洞上傳）3.黑帽seo 白帽seo 客戶網站被黑帽植入劫持代碼如何處理4.熟悉html表單 <html><head><title>TDK標題，描述 ......

    uj5u.com 2020-09-10 02:04:36 more
最新发布
  • 2023年最新微信小程式抓包教程

    01 開門見山 隔一個月發一篇文章，不過分。 首先回顧一下《微信系結手機號資料庫被脫庫事件》，我也是第一時間得知了這個訊息，然后跟蹤了整件事情的經過。下面是這起事件的相關截圖以及近日流出的一萬條資料樣本： 個人認為這件事也沒什么，還不如關注一下之前45億快遞資料查詢渠道疑似在近日復活的訊息。 訊息是 ......

    uj5u.com 2023-04-20 08:48:24 more
  • web3 產品介紹：metamask 錢包 使用最多的瀏覽器插件錢包

    Metamask錢包是一種基于區塊鏈技術的數字貨幣錢包，它允許用戶在安全、便捷的環境下管理自己的加密資產。Metamask錢包是以太坊生態系統中最流行的錢包之一，它具有易于使用、安全性高和功能強大等優點。 本文將詳細介紹Metamask錢包的功能和使用方法。 一、 Metamask錢包的功能 數字資 ......

    uj5u.com 2023-04-20 08:47:46 more
  • vulnhub_Earth

    前言 靶機地址->>>vulnhub_Earth 攻擊機ip：192.168.20.121 靶機ip：192.168.20.122 參考文章 https://www.cnblogs.com/Jing-X/archive/2022/04/03/16097695.html https://www.cnb ......

    uj5u.com 2023-04-20 07:46:20 more
  • 從4k到42k，軟體測驗工程師的漲薪史，給我看哭了

    清明節一過，盲猜大家已經無心上班，在數著日子準備過五一，但一想到銀行卡里的余額……瞬間心情就不美麗了。最近，2023年高校畢業生就業調查顯示，本科畢業月平均起薪為5825元。調查一出，便有很多同學表示自己又被平均了。看著這一資料，不免讓人想到前不久中國青年報的一項調查：近六成大學生認為畢業10年內會 ......

    uj5u.com 2023-04-20 07:44:00 more
  • 最新版本 Stable Diffusion 開源 AI 繪畫工具之中文自動提詞篇

    🎈 標簽生成器 由于輸入正向提示詞 prompt 和反向提示詞 negative prompt 都是使用英文，所以對學習母語的我們非常不友好 使用網址：https://tinygeeker.github.io/p/ai-prompt-generator 這個網址是為了讓大家在使用 AI 繪畫的時候 ......

    uj5u.com 2023-04-20 07:43:36 more
  • 漫談前端自動化測驗演進之路及測驗工具分析

    隨著前端技術的不斷發展和應用程式的日益復雜，前端自動化測驗也在不斷演進。隨著 Web 應用程式變得越來越復雜，自動化測驗的需求也越來越高。如今，自動化測驗已經成為 Web 應用程式開發程序中不可或缺的一部分，它們可以幫助開發人員更快地發現和修復錯誤，提高應用程式的性能和可靠性。 ......

    uj5u.com 2023-04-20 07:43:16 more
  • CANN開發實踐：4個DVPP記憶體問題的典型案例解讀

    摘要：由于DVPP媒體資料處理功能對存放輸入、輸出資料的記憶體有更高的要求（例如，記憶體首地址128位元組對齊），因此需呼叫專用的記憶體申請介面，那么本期就分享幾個關于DVPP記憶體問題的典型案例，并給出原因分析及解決方法。 本文分享自華為云社區《FAQ_DVPP記憶體問題案例》，作者：昇騰CANN。 DVPP ......

    uj5u.com 2023-04-20 07:43:03 more
  • msf學習

    msf學習 以kali自帶的msf為例 一、msf核心模塊與功能 msf模塊都放在/usr/share/metasploit-framework/modules目錄下 1、auxiliary 輔助模塊，輔助滲透（埠掃描、登錄密碼爆破、漏洞驗證等） 2、encoders 編碼器模塊，主要包含各種編碼 ......

    uj5u.com 2023-04-20 07:42:59 more
  • Halcon軟體安裝與界面簡介

    1. 下載Halcon17版本到到本地 2. 雙擊安裝包后 3. 步驟如下 1.2 Halcon軟體安裝 界面分為四大塊 1. Halcon的五個助手 1) 影像采集助手：與相機連接，設定相機引數，采集影像 2) 標定助手：九點標定或是其它的標定，生成標定檔案及內參外參，可以將像素單位轉換為長度單位 ......

    uj5u.com 2023-04-20 07:42:17 more
  • 在MacOS下使用Unity3D開發游戲

    第一次發博客，先發一下我的游戲開發環境吧。 去年2月份買了一臺MacBookPro2021 M1pro(以下簡稱mbp)，這一年來一直在用mbp開發游戲。我大致分享一下我的開發工具以及使用體驗。 1、Unity 官網鏈接： https://unity.cn/releases 我一般使用的Apple ......

    uj5u.com 2023-04-20 07:40:19 more

Copyright © 2010-2020 有解無憂