Kafka

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

為什么使用消息隊(duì)列MQ

在高并發(fā)環(huán)境下，同步請(qǐng)求來(lái)不及處理會(huì)發(fā)生堵塞，從而觸發(fā)too many connection錯(cuò)誤，引發(fā)雪崩效應(yīng)。比如大量的請(qǐng)求并發(fā)訪問(wèn)數(shù)據(jù)庫(kù)，導(dǎo)致行鎖表鎖，最后請(qǐng)求線程會(huì)堆積過(guò)多

我們使用消息隊(duì)列，通過(guò)異步請(qǐng)求，緩解系統(tǒng)壓力，消息隊(duì)列經(jīng)常應(yīng)用于異步處理，流量削峰，應(yīng)用解耦，消息通訊等場(chǎng)景

當(dāng)前比較常見(jiàn)的 MQ 中間件有 ActiveMQ、RabbitMQ、RocketMQ、Kafka 等
?

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

• 解耦

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

• 可恢復(fù)性

系統(tǒng)的一部分組件失效時(shí)，不會(huì)影響到整個(gè)系統(tǒng)。消息隊(duì)列降低了進(jìn)程間的耦合度，所以即使一個(gè)處理消息的進(jìn)程掛了，加入隊(duì)列的消息仍然可以在系統(tǒng)恢復(fù)后被處理

• 緩沖

有助于控制和優(yōu)化數(shù)據(jù)流結(jié)果系統(tǒng)的速度，解決生產(chǎn)消息和消費(fèi)消息的處理速度不一致的情況

• 靈活性，峰值處理能力

在訪問(wèn)量劇增的情況下，應(yīng)用仍然需要繼續(xù)發(fā)揮作用，但是這樣的突發(fā)流量并不常見(jiàn)
如果為以能處理這類(lèi)峰值訪問(wèn)為標(biāo)準(zhǔn)來(lái)投入資源隨時(shí)待命無(wú)疑是巨大的浪費(fèi)
使用消息隊(duì)列能夠使關(guān)鍵組件頂住突發(fā)的訪問(wèn)壓力，而不會(huì)因?yàn)橥话l(fā)的超負(fù)荷的請(qǐng)求而完全崩潰
?

• 異步通信

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

想向隊(duì)列中放入多少消息就放多少，然后在需要的時(shí)候再去處理它們
?

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

• 點(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)

• 發(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ì)多的依賴(lài)關(guān)系，使得每當(dāng)一個(gè)對(duì)象（目對(duì)標(biāo)象）的狀態(tài)發(fā)生改變，則所有依賴(lài)于它的對(duì)象（觀察者對(duì)象）都會(huì)得到通知并自動(dòng)更新

Kafka 概述

基于 Zookeeper

Kafka 是最初由 Linkedin 公司開(kāi)發(fā)，是一個(gè)分布式、支持分區(qū)的（partition）、多副本的（replicar 協(xié)調(diào)的分布式消息中間件系統(tǒng)，它的最大的特性就是可以實(shí)時(shí)的處理大量數(shù)據(jù)以滿足各種需求場(chǎng)景，比如基于 hadoop 的批處理系統(tǒng)、低延遲的實(shí)時(shí)系統(tǒng)、Spark/Flink 流式處理引擎，nginx 訪問(wèn)日志，消息服務(wù)等等，用 scala 語(yǔ)言編寫(xiě)

Linkedin 于 2010 年貢獻(xiàn)給了 Apache 基金會(huì)并成為頂級(jí)開(kāi)源項(xiàng)目
?

Kafka 特性

• 高吞吐量、低延遲

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

• 可擴(kuò)展性

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

• 持久性、可靠性

消息被持久化到本地磁盤(pán)，并且支持?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í)讀寫(xiě)

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

• Broker ? ? 服務(wù)器

一臺(tái) kafka 服務(wù)器就是一個(gè) broker
一個(gè)集群由多個(gè) broker 組成，一個(gè) broker 可以容納多個(gè) topic

• Topic ? 主題

可以理解為一個(gè)隊(duì)列，生產(chǎn)者和消費(fèi)者面向的都是一個(gè) topic
類(lèi)似于數(shù)據(jù)庫(kù)的表名或者 ES 的 index
物理上不同 topic 的消息分開(kāi)存儲(chǔ)

• Partition ?分區(qū)

為了實(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ì)列末尾

分區(qū)的原因

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

基礎(chǔ)架構(gòu)

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

2、Leader

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

3、Follower

Follower 跟隨 Leader，所有寫(xiě)請(qǐng)求都通過(guò) Leader 路由，數(shù)據(jù)變更會(huì)廣播給所有 Follower
Follower 與 Leader 保持?jǐn)?shù)據(jù)同步。Follower 只負(fù)責(zé)備份，不負(fù)責(zé)數(shù)據(jù)的讀寫(xiě)。
如果 Leader 故障，則從 Follower 中選舉出一個(gè)新的 Leader
當(dāng) Follower 掛掉、卡住或者同步太慢，Leader 會(huì)把這個(gè) Follower 從 ISR
（Leader 維護(hù)的一個(gè)和 Leader 保持同步的 Follower 集合） 列表中刪除，重新創(chuàng)建一個(gè) Follower

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

5、Consumer

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

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

7、offset 偏移量

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

8、Zookeeper

Kafka 通過(guò) Zookeeper 來(lái)存儲(chǔ)集群的 meta 信息

由于 consumer 在消費(fèi)過(guò)程中可能會(huì)出現(xiàn)斷電宕機(jī)等故障，consumer 恢復(fù)后，
需要從故障前的位置繼續(xù)消費(fèi)，所以 consumer 需要實(shí)時(shí)記錄自己消費(fèi)到了哪個(gè) offset，
以便故障恢復(fù)后繼續(xù)消費(fèi)

部署 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.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

修改 Kafka 配置文件

broker.id=0    
#21行，broker的全局唯一編號(hào)，每個(gè)broker不能重復(fù)，因此要在其他機(jī)器上配置 
broker.id=1、broker.id=2

listeners=PLAINTEXT://192.168.10.17:9092    
#31行，指定監(jiān)聽(tīng)的IP和端口，如果修改每個(gè)broker的IP需區(qū)分開(kāi)來(lái)，也可保持默認(rèn)配置不用修改
num.network.threads=3    #42行，broker 處理網(wǎng)絡(luò)請(qǐng)求的線程數(shù)量，一般情況下不需要去修改
num.io.threads=8         #45行，用來(lái)處理磁盤(pán)IO的線程數(shù)量，數(shù)值應(yīng)該大于硬盤(pán)數(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行，用來(lái)恢復(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.54.10:2181,192.168.154.20:2181,192.168.154.30:2181    
#123行，配置連接Zookeeper集群地址

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

//修改環(huán)境變量
vim /etc/profile
export KAFKA_HOME=/usr/local/kafka
export PATH=$PATH:$KAFKA_HOME/bin

source /etc/profile

配置 Zookeeper 啟動(dòng)腳本

//設(shè)置開(kāi)機(jī)自啟
chmod +x /etc/init.d/kafka
chkconfig --add kafka

//分別啟動(dòng) Kafka
service kafka start

Kafka 命令行操作

創(chuàng)建topic

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

kafka-topics.sh --list --zookeeper
192.168.154.10:2181,192.168.154.20:2181,192.168.154.30:2181

查看某個(gè) topic 詳情

kafka-topics.sh  --describe --zookeeper 
192.168.154.10:2181,192.168.154.20:2181,192.168.154.30:2181

發(fā)布消息

kafka-console-producer.sh --broker-list
192.168.154.10:2181,192.168.154.20:2181,192.168.154.30:2181  --topic test

消費(fèi)消息

kafka-console-consumer.sh --bootstrap-server 
192.168.154.10:2181,192.168.154.20:2181,192.168.154.30:2181
--topic test --from-beginning

--from-beginning：會(huì)把主題中以往所有的數(shù)據(jù)都讀取出來(lái)文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-726053.html

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

kafka-topics.sh --zookeeper 
192.168.154.10:2181,192.168.154.20:2181,192.168.154.30:2181
--alter --topic test --partitions 6

刪除 topic

kafka-topics.sh --delete --zookeeper 
192.168.154.10:2181,192.168.154.20:2181,192.168.154.30:2181 --topic test

到了這里，關(guān)于消息隊(duì)列 Kafka的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！