[[406253]]

 一、認(rèn)識(shí)kafka

Kafka到底是個(gè)啥?用來干嘛的?

官方定義如下：

Kafka is used for building real-time data pipelines and streaming apps. It is horizontally scalable, fault-tolerant, wicked fast, and runs in production in thousands of companies.

翻譯過來，大致的意思就是，這是一個(gè)實(shí)時(shí)數(shù)據(jù)處理系統(tǒng)，可以橫向擴(kuò)展，并高可靠!

實(shí)時(shí)數(shù)據(jù)處理，從名字上看，很好理解，就是將數(shù)據(jù)進(jìn)行實(shí)時(shí)處理，在現(xiàn)在流行的微服務(wù)開發(fā)中，最常用實(shí)時(shí)數(shù)據(jù)處理平臺(tái)有 RabbitMQ、RocketMQ 等消息中間件。

這些中間件，最大的特點(diǎn)主要有兩個(gè)：

• 服務(wù)解耦
• 流量削峰

在早期的 web 應(yīng)用程序開發(fā)中，當(dāng)請(qǐng)求量突然上來了時(shí)候，我們會(huì)將要處理的數(shù)據(jù)推送到一個(gè)隊(duì)列通道中，然后另起一個(gè)線程來不斷輪訓(xùn)拉取隊(duì)列中的數(shù)據(jù)，從而加快程序的運(yùn)行效率。

但是隨著請(qǐng)求量不斷的增大，并且隊(duì)列通道的數(shù)據(jù)一致處于高負(fù)載，在這種情況下，應(yīng)用程序的內(nèi)存占用率會(huì)非常高，稍有不慎，會(huì)出現(xiàn)內(nèi)存不足，造成程序內(nèi)存溢出，從而導(dǎo)致服務(wù)不可用。

隨著業(yè)務(wù)量的不斷擴(kuò)張，在一個(gè)應(yīng)用程序內(nèi)，使用這種模式已然無法滿足需求，因此之后，就誕生了各種消息中間件，例如 ActiveMQ、RabbitMQ、RocketMQ等中間件。

采用這種模型，本質(zhì)就是將要推送的數(shù)據(jù)，不在存放在當(dāng)前應(yīng)用程序的內(nèi)存中，而是將數(shù)據(jù)存放到另一個(gè)專門負(fù)責(zé)數(shù)據(jù)處理的應(yīng)用程序中，從而實(shí)現(xiàn)服務(wù)解耦。

消息中間件：主要的職責(zé)就是保證能接受到消息，并將消息存儲(chǔ)到磁盤，即使其他服務(wù)都掛了，數(shù)據(jù)也不會(huì)丟失，同時(shí)還可以對(duì)數(shù)據(jù)消費(fèi)情況做好監(jiān)控工作。

應(yīng)用程序：只需要將消息推送到消息中間件，然后啟用一個(gè)線程來不斷從消息中間件中拉取數(shù)據(jù)，進(jìn)行消費(fèi)確認(rèn)即可!

引入消息中間件之后，整個(gè)服務(wù)開發(fā)會(huì)變得更加簡(jiǎn)單，各負(fù)其責(zé)。

Kafka 本質(zhì)其實(shí)也是消息中間件的一種，Kafka 出自于 LinkedIn 公司，與 2010 年開源到 github。

LinkedIn 的開發(fā)團(tuán)隊(duì)，為了解決數(shù)據(jù)管道問題，起初采用了 ActiveMQ 來進(jìn)行數(shù)據(jù)交換，大約是在 2010 年前后，那時(shí)的 ActiveMQ 還遠(yuǎn)遠(yuǎn)無法滿足 LinkedIn 對(duì)數(shù)據(jù)傳遞系統(tǒng)的要求，經(jīng)常由于各種缺陷而導(dǎo)致消息阻塞或者服務(wù)無法正常訪問，為了能夠解決這個(gè)問題，LinkedIn 決定研發(fā)自己的消息傳遞系統(tǒng)，Kafka 由此誕生。

在 LinkedIn 公司，Kafka 可以有效地處理每天數(shù)十億條消息的指標(biāo)和用戶活動(dòng)跟蹤，其強(qiáng)大的處理能力，已經(jīng)被業(yè)界所認(rèn)可，并成為大數(shù)據(jù)流水線的首選技術(shù)。

二、架構(gòu)介紹

先來看一張圖，下面這張圖就是 kafka 生產(chǎn)與消費(fèi)的核心架構(gòu)模型!

如果你看不懂這些概念沒關(guān)系，我會(huì)帶著大家一起梳理一遍!

• Producer：Producer 即生產(chǎn)者，消息的產(chǎn)生者，是消息的入口
• Broker：Broker 是 kafka 一個(gè)實(shí)例，每個(gè)服務(wù)器上有一個(gè)或多個(gè) kafka 的實(shí)例，簡(jiǎn)單的理解就是一臺(tái) kafka 服務(wù)器，kafka cluster表示集群的意思
• Topic：消息的主題，可以理解為消息隊(duì)列，kafka的數(shù)據(jù)就保存在topic。在每個(gè) broker 上都可以創(chuàng)建多個(gè) topic 。
• Partition：Topic的分區(qū)，每個(gè) topic 可以有多個(gè)分區(qū)，分區(qū)的作用是做負(fù)載，提高 kafka 的吞吐量。同一個(gè) topic 在不同的分區(qū)的數(shù)據(jù)是不重復(fù)的，partition 的表現(xiàn)形式就是一個(gè)一個(gè)的文件夾!
• Replication：每一個(gè)分區(qū)都有多個(gè)副本，副本的作用是做備胎，主分區(qū)(Leader)會(huì)將數(shù)據(jù)同步到從分區(qū)(Follower)。當(dāng)主分區(qū)(Leader)故障的時(shí)候會(huì)選擇一個(gè)備胎(Follower)上位，成為 Leader。在kafka中默認(rèn)副本的最大數(shù)量是10個(gè)，且副本的數(shù)量不能大于Broker的數(shù)量，follower和leader絕對(duì)是在不同的機(jī)器，同一機(jī)器對(duì)同一個(gè)分區(qū)也只可能存放一個(gè)副本
• Message：每一條發(fā)送的消息主體。
• Consumer：消費(fèi)者，即消息的消費(fèi)方，是消息的出口。
• Consumer Group：我們可以將多個(gè)消費(fèi)組組成一個(gè)消費(fèi)者組，在 kafka 的設(shè)計(jì)中同一個(gè)分區(qū)的數(shù)據(jù)只能被消費(fèi)者組中的某一個(gè)消費(fèi)者消費(fèi)。同一個(gè)消費(fèi)者組的消費(fèi)者可以消費(fèi)同一個(gè)topic的不同分區(qū)的數(shù)據(jù)，這也是為了提高kafka的吞吐量!
• Zookeeper：kafka 集群依賴 zookeeper 來保存集群的的元信息，來保證系統(tǒng)的可用性。

簡(jiǎn)而言之，kafka 本質(zhì)就是一個(gè)消息系統(tǒng)，與大多數(shù)的消息系統(tǒng)一樣，主要的特點(diǎn)如下：

• 使用推拉模型將生產(chǎn)者和消費(fèi)者分離
• 為消息傳遞系統(tǒng)中的消息數(shù)據(jù)提供持久性，以允許多個(gè)消費(fèi)者
• 提供高可用集群服務(wù)，主從模式，同時(shí)支持橫向水平擴(kuò)展

與 ActiveMQ、RabbitMQ、RocketMQ 不同的地方在于，它有一個(gè)**分區(qū)Partition**的概念。

這個(gè)分區(qū)的意思就是說，如果你創(chuàng)建的topic有5個(gè)分區(qū)，當(dāng)你一次性向 kafka 中推 1000 條數(shù)據(jù)時(shí)，這 1000 條數(shù)據(jù)默認(rèn)會(huì)分配到 5 個(gè)分區(qū)中，其中每個(gè)分區(qū)存儲(chǔ) 200 條數(shù)據(jù)。

這樣做的目的，就是方便消費(fèi)者從不同的分區(qū)拉取數(shù)據(jù)，假如你啟動(dòng) 5 個(gè)線程同時(shí)拉取數(shù)據(jù)，每個(gè)線程拉取一個(gè)分區(qū)，消費(fèi)速度會(huì)非常非常快!

這是 kafka 與其他的消息系統(tǒng)最大的不同!

2.1、發(fā)送數(shù)據(jù)

和其他的中間件一樣，kafka 每次發(fā)送數(shù)據(jù)都是向Leader分區(qū)發(fā)送數(shù)據(jù)，并順序?qū)懭氲酱疟P，然后Leader分區(qū)會(huì)將數(shù)據(jù)同步到各個(gè)從分區(qū)Follower，即使主分區(qū)掛了，也不會(huì)影響服務(wù)的正常運(yùn)行。

那 kafka 是如何將數(shù)據(jù)寫入到對(duì)應(yīng)的分區(qū)呢?kafka中有以下幾個(gè)原則：

1、數(shù)據(jù)在寫入的時(shí)候可以指定需要寫入的分區(qū)，如果有指定，則寫入對(duì)應(yīng)的分區(qū)

2、如果沒有指定分區(qū)，但是設(shè)置了數(shù)據(jù)的key，則會(huì)根據(jù)key的值hash出一個(gè)分區(qū)

3、如果既沒指定分區(qū)，又沒有設(shè)置key，則會(huì)輪詢選出一個(gè)分區(qū)

2.2、消費(fèi)數(shù)據(jù)

與生產(chǎn)者一樣，消費(fèi)者主動(dòng)的去kafka集群拉取消息時(shí)，也是從Leader分區(qū)去拉取數(shù)據(jù)。

這里我們需要重點(diǎn)了解一個(gè)名詞：消費(fèi)組!

考慮到多個(gè)消費(fèi)者的場(chǎng)景，kafka 在設(shè)計(jì)的時(shí)候，可以由多個(gè)消費(fèi)者組成一個(gè)消費(fèi)組，同一個(gè)消費(fèi)組者的消費(fèi)者可以消費(fèi)同一個(gè) topic 下不同分區(qū)的數(shù)據(jù)，同一個(gè)分區(qū)只會(huì)被一個(gè)消費(fèi)組內(nèi)的某個(gè)消費(fèi)者所消費(fèi)，防止出現(xiàn)重復(fù)消費(fèi)的問題!

但是不同的組，可以消費(fèi)同一個(gè)分區(qū)的數(shù)據(jù)!

你可以這樣理解，一個(gè)消費(fèi)組就是一個(gè)客戶端，一個(gè)客戶端可以由很多個(gè)消費(fèi)者組成，以便加快消息的消費(fèi)能力。

但是，如果一個(gè)組下的消費(fèi)者數(shù)量大于分區(qū)數(shù)量，就會(huì)出現(xiàn)很多的消費(fèi)者閑置。

如果分區(qū)數(shù)量大于一個(gè)組下的消費(fèi)者數(shù)量，會(huì)出現(xiàn)一個(gè)消費(fèi)者負(fù)責(zé)多個(gè)分區(qū)的消費(fèi)，會(huì)出現(xiàn)消費(fèi)性能不均衡的情況。

因此，在實(shí)際的應(yīng)用中，建議消費(fèi)者組的consumer的數(shù)量與partition的數(shù)量保持一致!

三、kafka 安裝

光說理論可沒用，下面我們就以 centos7 為例，介紹一下 kafka 的安裝和使用。

kafka 需要 zookeeper 來保存服務(wù)實(shí)例的元信息，因此在安裝 kafka 之前，我們需要先安裝 zookeeper。

3.1、安裝zookeeper

zookeeper 安裝環(huán)境依賴于 jdk，因此我們需要事先安裝 jdk

# 安裝jdk1.8 
yum -y install java-1.8.0-openjdk 

下載zookeeper，并解壓文件包

#在線下載zookeeper 
wget http://mirrors.hust.edu.cn/apache/zookeeper/zookeeper-3.4.12/zookeeper-3.4.12.tar.gz 
 
#解壓 
tar -zxvf zookeeper-3.4.12.tar.gz 

創(chuàng)建數(shù)據(jù)、日志目錄

#創(chuàng)建數(shù)據(jù)和日志存放目錄 
cd /usr/zookeeper/ 
mkdir data 
mkdir log 
 
#把conf下的zoo_sample.cfg備份一份，然后重命名為zoo.cfg 
cd conf/ 
cp zoo_sample.cfg zoo.cfg 

配置zookeeper

#編輯zoo.cfg文件 
vim zoo.cfg 

重新配置dataDir和dataLogDir的存儲(chǔ)路徑

最后，啟動(dòng) Zookeeper 服務(wù)

#進(jìn)入Zookeeper的bin目錄 
cd zookeeper/zookeeper-3.4.12/bin 
 
#啟動(dòng)Zookeeper 
./zkServer.sh start 
 
#查詢Zookeeper狀態(tài) 
./zkServer.sh status 
 
#關(guān)閉Zookeeper狀態(tài) 
./zkServer.sh stop 

3.2、安裝kafka

到官網(wǎng)http://kafka.apache.org/downloads.html下載想要的版本，我這里下載是最新穩(wěn)定版2.8.0。

#下載kafka 安裝包 
wget https://apache.osuosl.org/kafka/2.8.0/kafka-2.8.0-src.tgz 
 
#解壓文件包 
tar -xvf kafka-2.8.0-src.tgz 

按需修改配置文件server.properties(可選)

#進(jìn)入配置文件夾 
cd  kafka-2.8.0-src/config 
 
#編輯server.properties 
vim server.properties 

server.properties文件內(nèi)容如下：

broker.id=0 
listeners=PLAINTEXT://localhost:9092 
num.network.threads=3 
num.io.threads=8 
socket.send.buffer.bytes=102400 
socket.receive.buffer.bytes=102400 
socket.request.max.bytes=104857600 
log.dirs=/tmp/kafka-logs 
num.partitions=1 
num.recovery.threads.per.data.dir=1 
offsets.topic.replication.factor=1 
transaction.state.log.replication.factor=1 
transaction.state.log.min.isr=1 
log.retention.hours=168 
log.segment.bytes=1073741824 
log.retention.check.interval.ms=300000 
zookeeper.connect=localhost:2181 
zookeeper.connection.timeout.ms=6000 
group.initial.rebalance.delay.ms=0 

其中有四個(gè)重要的參數(shù)：

• broker.id：唯一標(biāo)識(shí)ID
• listeners=PLAINTEXT://localhost:9092：kafka服務(wù)監(jiān)聽地址和端口
• log.dirs：日志存儲(chǔ)目錄
• zookeeper.connect：指定zookeeper服務(wù)地址

可根據(jù)自己需求修改對(duì)應(yīng)的配置!

3.3、啟動(dòng) kafka 服務(wù)

# 進(jìn)入bin腳本目錄 
cd  kafka-2.8.0-src/bin 

啟動(dòng) kafka 服務(wù)

nohup kafka-server-start.sh ../config/server.properties server.log 2> server.err & 

3.4、創(chuàng)建主題topics

創(chuàng)建一個(gè)名為testTopic的主題，它只包含一個(gè)分區(qū)，只有一個(gè)副本：

# 進(jìn)入bin腳本目錄 
cd  kafka-2.8.0-src/bin 
 
#創(chuàng)建topics 
kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic testTopic 

運(yùn)行l(wèi)ist topic命令，可以看到該主題。

# 進(jìn)入bin腳本目錄 
cd  kafka-2.8.0-src/bin 
 
#查詢當(dāng)前kafka上所有的主題 
kafka-topics.sh --list --zookeeper localhost:2181 

輸出內(nèi)容：

testTopic 

3.5、發(fā)送消息

Kafka 附帶一個(gè)命令行客戶端，它將從文件或標(biāo)準(zhǔn)輸入中獲取輸入，并將其作為消息發(fā)送到 Kafka 集群。默認(rèn)情況下，每行將作為單獨(dú)的消息發(fā)送。

運(yùn)行生產(chǎn)者，然后在控制臺(tái)中鍵入一些消息以發(fā)送到服務(wù)器。

# 進(jìn)入bin腳本目錄 
cd  kafka-2.8.0-src/bin 
 
#運(yùn)行一個(gè)生產(chǎn)者，向testTopic主題中發(fā)消息 
kafka-console-producer.sh --broker-list localhost:9092 --topic testTopic 

輸入兩條內(nèi)容并回車：

Hello kafka! 
This is a message 

3.5、接受消息

Kafka 還有一個(gè)命令行使用者，它會(huì)將消息轉(zhuǎn)儲(chǔ)到標(biāo)準(zhǔn)輸出。

# 進(jìn)入bin腳本目錄 
cd  kafka-2.8.0-src/bin 
 
#運(yùn)行一個(gè)消費(fèi)者，從testTopic主題中拉取消息 
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic testTopic --from-beginning 

輸出結(jié)果如下：

Hello kafka! 
This is a message 

四、小結(jié)

本文主要圍繞 kafka 的架構(gòu)模型和安裝環(huán)境做了一些初步的介紹，難免會(huì)有理解不對(duì)的地方，歡迎網(wǎng)友批評(píng)、吐槽。

由于篇幅原因，會(huì)在下期文章中詳細(xì)介紹 java 環(huán)境下 kafka 應(yīng)用場(chǎng)景!