一、kafka概述

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

• 主要原因是由于在高并發(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)的壓力。消息隊(duì)列常應(yīng)用于異步處理，流量削峰，應(yīng)用解耦，消息通訊等場(chǎng)景。

1.2常見的中間

消息隊(duì)列型（MQ）

• ActiveMQ、RabbitMQ、RocketMQ、Kafka、Pulsar、Redis

Web應(yīng)用型（代理服務(wù)器）

• Nginx、Haproxy、LVS、Tomcat、php

1.3消息隊(duì)列的優(yōu)點(diǎn)

（1）解耦

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

（2）可恢復(fù)性

• 系統(tǒng)的一部分組件失效時(shí)，不會(huì)影響到整個(gè)系統(tǒng)。消息隊(duì)列降低了進(jìn)程間的耦合度，所以即使一個(gè)處理消息的進(jìn)程掛掉，加入隊(duì)列中的消息仍然可以在系統(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)來投入資源隨時(shí)待命無疑是巨大的浪費(fèi)。使用消息隊(duì)列能夠使關(guān)鍵組件頂住突發(fā)的訪問壓力，而不會(huì)因?yàn)橥话l(fā)的超負(fù)荷的請(qǐng)求而完全崩潰。

（5）異步通信

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

1.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)更新。

1.5 Kafka 定義

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

1.6 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)失敗（多副本情況下，若副本數(shù)量為 n，則允許 n-1 個(gè)節(jié)點(diǎn)失敗）

●高并發(fā)

• 支持?jǐn)?shù)千個(gè)客戶端同時(shí)讀寫

1.7kafka的系統(tǒng)架構(gòu)

（1）Broker

• 一臺(tái) kafka 服務(wù)器就是一個(gè) broker。一個(gè)集群由多個(gè) broker 組成。一個(gè) broker 可以容納多個(gè) topic。
  （2）Topic
• 可以理解為一個(gè)隊(duì)列，生產(chǎn)者和消費(fèi)者面向的都是一個(gè) topic。
  類似于數(shù)據(jù)庫(kù)的表名或者 ES 的 index
  物理上不同 topic 的消息分開存儲(chǔ)

（3）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ì)列末尾。
  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)榭梢砸訮artition為單位讀寫了。

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

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

（6）Follower
Follower 跟隨 Leader，所有寫請(qǐng)求都通過 Leader 路由，數(shù)據(jù)變更會(huì)廣播給所有 Follower，F(xiàn)ollower 與 Leader 保持?jǐn)?shù)據(jù)同步。Follower 只負(fù)責(zé)備份，不負(fù)責(zé)數(shù)據(jù)的讀寫。
如果 Leader 故障，則從 Follower 中選舉出一個(gè)新的 Leader。
當(dāng) Follower 掛掉、卡住或者同步太慢，Leader 會(huì)把這個(gè) Follower 從 ISR（Leader 維護(hù)的一個(gè)和 Leader 保持同步的 Follower 集合） 列表中刪除，重新創(chuàng)建一個(gè) 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ā)送的消息，存儲(chǔ)到一個(gè) partition 中，生產(chǎn)者也可以指定數(shù)據(jù)存儲(chǔ)的 partition。

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

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

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

（11）Zookeeper
Kafka 通過 Zookeeper 來存儲(chǔ)集群的 meta 信息。

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

2.1安裝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.7.1.tgz
mv kafka_2.13-2.7.1 /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.80.10: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.80.10:2181,192.168.80.11:2181,192.168.80.12:2181    ●123行，配置連接Zookeeper集群地址

//修改環(huán)境變量

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


source /etc/profile

//配置 Zookeeper 啟動(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

2.2Kafka 命令行操作

創(chuàng)建topic

kafka-topics.sh --create --zookeeper 192.168.243.105:2181,192.168.243.106:2181,192.168.243.107:2181 --replication-factor 2 --partitions 3 --topic dodo

• –zookeeper：定義 zookeeper 集群服務(wù)器地址，如果有多個(gè) IP 地址使用逗號(hào)分割，一般使用一個(gè) IP 即可
• –replication-factor：定義分區(qū)副本數(shù)，1 代表單副本，建議為 2
• –partitions：定義分區(qū)數(shù)
• –topic：定義 topic 名稱

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

afka-topics.sh --list --zookeeper 192.168.243.105:2181

查看某個(gè) topic 的詳情

 kafka-topics.sh --describe --zookeeper 192.168.243.105:2181 --topic dodo

發(fā)布消息

kafka-console-producer.sh --broker-list 192.168.243.105:9092,192.168.243.106:9092,192.168.243.107:9092 --topic dodo

消費(fèi)消息

/kafka-console-consumer.sh --bootstrap-server 192.168.243.105:9092,192.168.243.106:9092,192.168.243.107:9092 --topic dodo --from-beginning

–from-beginning：會(huì)把主題中以往所有的數(shù)據(jù)都讀取出來

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

afka-topics.sh --describe --zookeeper 192.168.243.105:2181 --topic dodo

刪除 topic

kafka-topics.sh --delete --zookeeper 192.168.243.105:2181,192.168.243.106:2191,192.168.243.107:2181 --topic tes

三、kafka架構(gòu)深入

3.1kfka工作流程及文件存儲(chǔ)機(jī)制

• Kafka 中消息是以 topic 進(jìn)行分類的，生產(chǎn)者生產(chǎn)消息，消費(fèi)者消費(fèi)消息，都是面向 topic 的。
• topic 是邏輯上的概念，而 partition 是物理上的概念，每個(gè) partition 對(duì)應(yīng)于一個(gè) log 文件，該 log 文件中存儲(chǔ)的就是 producer 生產(chǎn)的數(shù)據(jù)。Producer 生產(chǎn)的數(shù)據(jù)會(huì)被不斷追加到該 log 文件末端，且每條數(shù)據(jù)都有自己的 offset。 消費(fèi)者組中的每個(gè)消費(fèi)者，都會(huì)實(shí)時(shí)記錄自己消費(fèi)到了哪個(gè) offset，以便出錯(cuò)恢復(fù)時(shí)，從上次的位置繼續(xù)消費(fèi)。
• 由于生產(chǎn)者生產(chǎn)的消息會(huì)不斷追加到 log 文件末尾，為防止 log 文件過大導(dǎo)致數(shù)據(jù)定位效率低下，Kafka 采取了分片和索引機(jī)制，將每個(gè) partition 分為多個(gè) segment。每個(gè) segment 對(duì)應(yīng)兩個(gè)文件：“.index” 文件和 “.log” 文件。這些文件位于一個(gè)文件夾下，該文件夾的命名規(guī)則為：topic名稱+分區(qū)序號(hào)。例如，test 這個(gè) topic 有三個(gè)分區(qū)， 則其對(duì)應(yīng)的文件夾為 test-0、test-1、test-2。
• index 和 log 文件以當(dāng)前 segment 的第一條消息的 offset 命名。
• “.index” 文件存儲(chǔ)大量的索引信息，“.log” 文件存儲(chǔ)大量的數(shù)據(jù)，索引文件中的元數(shù)據(jù)指向?qū)?yīng)數(shù)據(jù)文件中 message 的物理偏移地址。

數(shù)據(jù)可靠性保證

為保證 producer 發(fā)送的數(shù)據(jù)，能可靠的發(fā)送到指定的 topic，topic 的每個(gè) partition 收到 producer 發(fā)送的數(shù)據(jù)后， 都需要向 producer 發(fā)送 ack（acknowledgement 確認(rèn)收到），如果 producer 收到 ack，就會(huì)進(jìn)行下一輪的發(fā)送，否則重新發(fā)送數(shù)據(jù)。

3.2數(shù)據(jù)一致性問題

• LEO：指的是每個(gè)副本最大的 offset
• HW：指的是消費(fèi)者能見到的最大的 offset，所有副本中最小的 LEO。
  （1）follower 故障

follower 發(fā)生故障后會(huì)被臨時(shí)踢出 ISR（Leader 維護(hù)的一個(gè)和 Leader 保持同步的 Follower 集合），待該 follower 恢復(fù)后，follower 會(huì)讀取本地磁盤記錄的上次的 HW，并將 log 文件高于 HW 的部分截取掉，從 HW 開始向 leader 進(jìn)行同步。等該 follower 的 LEO 大于等于該 Partition 的 HW，即 follower 追上 leader 之后，就可以重新加入 ISR 了。

（2）leader 故障

leader 發(fā)生故障之后，會(huì)從 ISR 中選出一個(gè)新的 leader， 之后，為保證多個(gè)副本之間的數(shù)據(jù)一致性，其余的 follower會(huì)先將各自的 log 文件高于 HW 的部分截掉，然后從新的 leader 同步數(shù)據(jù)。

注：這只能保證副本之間的數(shù)據(jù)一致性，并不能保證數(shù)據(jù)不丟失或者不重復(fù)。

3.3ack應(yīng)答機(jī)制

• 對(duì)于某些不太重要的數(shù)據(jù)，對(duì)數(shù)據(jù)的可靠性要求不是很高，能夠容忍數(shù)據(jù)的少量丟失，所以沒必要等 ISR 中的 follower 全部接收成功。所以 Kafka 為用戶提供了三種可靠性級(jí)別，用戶根據(jù)對(duì)可靠性和延遲的要求進(jìn)行權(quán)衡選擇。

• 當(dāng) producer 向 leader 發(fā)送數(shù)據(jù)時(shí)，可以通過 request.required.acks 參數(shù)來設(shè)置數(shù)據(jù)可靠性的級(jí)別：

●0：這意味著producer無需等待來自broker的確認(rèn)而繼續(xù)發(fā)送下一批消息。這種情況下數(shù)據(jù)傳輸效率最高，但是數(shù)據(jù)可靠性確是最低的。當(dāng)broker故障時(shí)有可能丟失數(shù)據(jù)。

●1（默認(rèn)配置）：這意味著producer在ISR中的leader已成功收到的數(shù)據(jù)并得到確認(rèn)后發(fā)送下一條message。如果在follower同步成功之前l(fā)eader故障，那么將會(huì)丟失數(shù)據(jù)。

●-1（或者是all）：producer需要等待ISR中的所有follower都確認(rèn)接收到數(shù)據(jù)后才算一次發(fā)送完成，可靠性最高。但是如果在
follower 同步完成后，broker 發(fā)送ack 之前，leader 發(fā)生故障，那么會(huì)造成數(shù)據(jù)重復(fù)。
三種機(jī)制性能依次遞減，數(shù)據(jù)可靠性依次遞增。

• 注：在 0.11 版本以前的Kafka，對(duì)此是無能為力的，只能保證數(shù)據(jù)不丟失，再在下游消費(fèi)者對(duì)數(shù)據(jù)做全局去重。在 0.11 及以后版本的 Kafka，引入了一項(xiàng)重大特性：冪等性。所謂的冪等性就是指 Producer 不論向 Server 發(fā)送多少次重復(fù)數(shù)據(jù)， Server 端都只會(huì)持久化一條。

四、Filebeat+Kafka+ELK

4.1部署 Zookeeper+Kafka 集群

Zookeeper集群搭建
kafka部署看上文

4.2部署 Filebeat

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.80.10:9092","192.168.80.11:9092","192.168.80.12:9092"]    #指定 Kafka 集群配置
  topic: "httpd"    #指定 Kafka 的 topic
  

啟動(dòng) filebeat

./filebeat -e -c filebeat.yml

4.3部署 ELK，在 Logstash 組件所在節(jié)點(diǎn)上新建一個(gè) Logstash 配置文件

cd /etc/logstash/conf.d/

vim kafka.conf
input {
    kafka {
        bootstrap_servers => "192.168.80.10:9092,192.168.80.11:9092,192.168.80.12: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.80.30:9200"]
      index => "httpd_access-%{+YYYY.MM.dd}"
    }
  }
  
  if "error" in [tags] {
    elasticsearch {
      hosts => ["192.168.80.30:9200"]
      index => "httpd_error-%{+YYYY.MM.dd}"
    }
  }
  
  stdout { codec => rubydebug }
}

啟動(dòng) logstash

logstash -f kafka.conf

4.4瀏覽器訪問驗(yàn)證

http://192.168.243.100:5601 登錄 Kibana，單擊“Create Index Pattern”按鈕添加索引“filebeat_test-*”，單擊 “create” 按鈕創(chuàng)建，單擊 “Discover” 按鈕可查看圖表信息及日志信息。

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

到了這里，關(guān)于【分布式應(yīng)用】kafka集群、Filebeat+Kafka+ELK搭建的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！