昨日內(nèi)容回顧:
?? ?- ES的加密及客戶端的連接方式，比如logstash，filebeat，curl，es-head，postman...
?? ?- zk單點(diǎn)部署
?? ?- kafka單點(diǎn)部署
?? ?- kafka的集群擴(kuò)容
?? ?- kafka集群架構(gòu)
?? ?- kafka的堆內(nèi)存調(diào)優(yōu)
?? ?- kafka的集群宏觀架構(gòu)
?? ?

Q1: 為什么需要配置"dvertised.listeners".
?? ?因?yàn)闆]有配置主機(jī)名解析導(dǎo)致的問題。
?? ?
Q2: 一個(gè)分區(qū)數(shù)寫入數(shù)據(jù)是有序的，但為什么多個(gè)分區(qū)數(shù)寫入就是無序的呢？
?? ?因?yàn)橄M(fèi)者是從多個(gè)分區(qū)讀取的，順序無法確定。
?? ?
Q3: leader同步follower是如何同步的？
?? ?follower主動(dòng)去leader拉取數(shù)據(jù)。

Q4: leader掛掉之后發(fā)生了哪些事，請(qǐng)說明？
?? ?leader掛掉后，follower會(huì)成為新的leader，需要借助zookeeper實(shí)現(xiàn)。

Q5: follower掛掉之后如何解決，會(huì)發(fā)生哪些問題？
?? ?leader會(huì)根據(jù)replica.lag.time.max.ms定義的間隔時(shí)間，超出該范圍leader就會(huì)將其剔除ISR列表。

Q6: 當(dāng)follower數(shù)據(jù)和leader不同時(shí)，又會(huì)發(fā)生啥事？
?? ?leader會(huì)根據(jù)replica.lag.time.max.ms定義的間隔時(shí)間，超出該范圍leader就會(huì)將其剔除ISR列表。

Q7: 當(dāng)消費(fèi)者數(shù)據(jù)延遲時(shí)如何解決問題？請(qǐng)說明原因。
?? ?將多個(gè)消費(fèi)者加入到同一個(gè)消費(fèi)者組進(jìn)行消費(fèi)，但是要注意消費(fèi)者數(shù)量不能大于分區(qū)數(shù)數(shù)量。

Q8: 請(qǐng)問多副本，例如一個(gè)分區(qū)3個(gè)副本，kafka會(huì)存在數(shù)據(jù)丟失的風(fēng)險(xiǎn)嗎？請(qǐng)說明原因。
?? ?會(huì)，還是ISR的存在。

相關(guān)術(shù)語:
?? ?Log End Offset (簡稱"LEO"):
?? ??? ?每個(gè)partition的最后一個(gè)日志的偏移量。
?? ??? ?
?? ?High Water(簡稱"HW"):
?? ??? ?所有ISR列表中最小的LEO。
?? ??? ?所有的消費(fèi)者只能讀取到HW之前的偏移量數(shù)據(jù)。
?? ??? ?
?? ?ISR:
?? ??? ?和leader數(shù)據(jù)同步的所有副本集合。
?? ??? ?
?? ?OSR:
?? ??? ?和leader數(shù)據(jù)不同的所有副本集合。

?? ?AR:
?? ??? ?是ISR+OSR，指的是所有副本集合。

?? ?replica.lag.time.max.ms: 30000ms ?---> 30s
?? ??? ?當(dāng)follower節(jié)點(diǎn)超過30秒沒有向leader發(fā)送fetch請(qǐng)求或者follower的LEO不等于leader的LEO時(shí)，leader會(huì)將follower踢出ISR列表。
?? ??? ?
?? ??? ?

zookeeper的基礎(chǔ)操作:
?? ?1.簡介
zookeeper是一個(gè)分布式協(xié)調(diào)服務(wù)，其扮演的是輔助的功能。本身并不支持存儲(chǔ)大量數(shù)據(jù)，每個(gè)znode默認(rèn)可以存儲(chǔ)大約2M的數(shù)據(jù)。

?? ?2.什么是znode
所謂的znode指的是zookeeper node，相當(dāng)于Linux的目錄，只不過znode下不僅僅可以存儲(chǔ)子znode，其本身也能夠存數(shù)據(jù)。

?? ?3.znode的基礎(chǔ)管理
登錄:
# zkCli.sh?
增:
?? ?create /test
?? ??? ?創(chuàng)建一個(gè)"/test"的znode。
?? ?
?? ?create /test/b
?? ??? ?支持多級(jí)路徑創(chuàng)建，但要求父路徑是必須存在的。
?? ??? ?
?? ?

刪:
?? ?delete /test/a?
?? ??? ?刪除"/test/a"的znode，要求znode為空。
?? ??? ?
?? ?deleteall /test?
?? ??? ?刪除"/test"的所有內(nèi)容，可以刪除非空znode。

改:
?? ?set /test/c/1111 bbbbbbbbbb
?? ??? ?修改"/test/c/1111"的znode存儲(chǔ)數(shù)據(jù)。
?? ??? ?
?? ?
查: ? ? ? ? ? ?*****
?? ?ls /
?? ??? ?查看/的znode下有哪些子znode。
?? ?
?? ?ls /test -R
?? ??? ?遞歸查看"/test"子路徑下的所有znode。
?? ??? ?
?? ?get /test
?? ??? ?查看數(shù)據(jù)。
?? ??? ?
?? ??? ?
?? ??? ?
kafka 0.8.0 多實(shí)例部署:
?? ?1.解壓軟包包
tar xf kafka_2.8.0-0.8.0.tar.gz ?-C /oldboyedu/softwares/
?
?? ?2.修改配置文件
vim /oldboyedu/softwares/kafka_2.8.0-0.8.0/config/server.properties
...
broker.id=101
# kafka的監(jiān)聽端口
port=19092
log.dirs=/oldboyedu/data/kafka080
zookeeper.connect=10.0.0.101:2181/oldboyedu-linux82-kafka080

?? ?3.101節(jié)點(diǎn)同步數(shù)據(jù)
data_rsync.sh /oldboyedu/softwares/kafka_2.8.0-0.8.0/?? ?
?? ?
?? ?4.102節(jié)點(diǎn)修改broker.id
vim /oldboyedu/softwares/kafka_2.8.0-0.8.0/config/server.properties
...
broker.id=102
?
?? ?5.103節(jié)點(diǎn)修改broker.id
vim /oldboyedu/softwares/kafka_2.8.0-0.8.0/config/server.properties
...
broker.id=102

?? ?6.zookeeper創(chuàng)建znode
create /oldboyedu-linux82-kafka080

?? ?
?? ?7.啟動(dòng)kafka
cd /oldboyedu/softwares/kafka_2.8.0-0.8.0 ?&& ./bin/kafka-server-start.sh config/server.properties &>/dev/null &

?? ?8.檢查zookeeper集群的ids信息
ls /oldboyedu-linux82-kafka080/brokers/ids

早期kafka 0.9之前的offset存儲(chǔ)在zookeeper：
?? ?1.查看topic信息
./bin/kafka-list-topic.sh --zookeeper 10.0.0.101:2181/oldboyedu-linux82-kafka080
?? ??? ?
?? ?2.啟動(dòng)生產(chǎn)者
./bin/kafka-console-producer.sh ?--broker-list 10.0.0.101:19092,10.0.0.102:19092,10.0.0.103:19092 --topic oldboyedu-linux82

?? ?3.啟動(dòng)消費(fèi)者
./bin/kafka-console-consumer.sh --zookeeper 10.0.0.101:2181/oldboyedu-linux82-kafka080 --topic oldboyedu-linux82

?? ?4.查看偏移量，注意"console-consumer-78441"根據(jù)您的環(huán)境適當(dāng)改變即可
get /oldboyedu-linux82-kafka080/consumers/console-consumer-78441/offsets/oldboyedu-linux82/0

從kafka 0.9之后的版本之后支持將offset存儲(chǔ)在kafka的"__consumer_offsets"內(nèi)置topic中：
?? ?1.查看kafka內(nèi)置"__consumer_offsets"的方式
kafka-console-consumer.sh --bootstrap-server 10.0.0.101:9092 ?--topic __consumer_offsets ?--formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" --from-beginning ?| grep oldboyedu-linux82

....

# [消費(fèi)者組,消費(fèi)的topic,分區(qū)數(shù)]::OffsetAndMetadata(偏移量，...)
[linux82-elk,oldboyedu-linux82,4]::OffsetAndMetadata(offset=5, leaderEpoch=Optional[0], metadata=, commitTimestamp=1661498621467, expireTimestamp=None)
[linux82-elk,oldboyedu-linux82,1]::OffsetAndMetadata(offset=4, leaderEpoch=Optional[0], metadata=, commitTimestamp=1661498621467, expireTimestamp=None)
[linux82-elk,oldboyedu-linux82,2]::OffsetAndMetadata(offset=3, leaderEpoch=Optional[0], metadata=, commitTimestamp=1661498621467, expireTimestamp=None)
[linux82-elk,oldboyedu-linux82,0]::OffsetAndMetadata(offset=4, leaderEpoch=Optional[0], metadata=, commitTimestamp=1661498621467, expireTimestamp=None)
[linux82-elk,oldboyedu-linux82,3]::OffsetAndMetadata(offset=3, leaderEpoch=Optional[0], metadata=, commitTimestamp=1661498621467, expireTimestamp=None)

?? ?2.查看消費(fèi)者組信息，注意觀察"linux82-elk"的消費(fèi)者組是否有消息延遲！
kafka-consumer-groups.sh --bootstrap-server 10.0.0.101:9092 ?--describe --all-groups
kafka-consumer-groups.sh --bootstrap-server 10.0.0.101:9092 ?--describe --group linux82-elk

...
GROUP ? ? ? ? ? TOPIC ? ? ? ? ? ? PARTITION ?CURRENT-OFFSET ?LOG-END-OFFSET ?LAG ? ? ? ? ? ? CONSUMER-ID ? ? HOST ? ? ? ? ? ?CLIENT-ID
linux82-elk ? ? oldboyedu-linux82 4 ? ? ? ? ?5 ? ? ? ? ? ? ? 6 ? ? ? ? ? ? ? 1 ? ? ? ? ? ? ? - ? ? ? ? ? ? ? - ? ? ? ? ? ? ? -
linux82-elk ? ? oldboyedu-linux82 1 ? ? ? ? ?4 ? ? ? ? ? ? ? 4 ? ? ? ? ? ? ? 0 ? ? ? ? ? ? ? - ? ? ? ? ? ? ? - ? ? ? ? ? ? ? -
linux82-elk ? ? oldboyedu-linux82 2 ? ? ? ? ?3 ? ? ? ? ? ? ? 4 ? ? ? ? ? ? ? 1 ? ? ? ? ? ? ? - ? ? ? ? ? ? ? - ? ? ? ? ? ? ? -
linux82-elk ? ? oldboyedu-linux82 0 ? ? ? ? ?4 ? ? ? ? ? ? ? 5 ? ? ? ? ? ? ? 1 ? ? ? ? ? ? ? - ? ? ? ? ? ? ? - ? ? ? ? ? ? ? -
linux82-elk ? ? oldboyedu-linux82 3 ? ? ? ? ?3 ? ? ? ? ? ? ? 5 ? ? ? ? ? ? ? 2 ? ? ? ? ? ? ? - ? ? ? ? ? ? ? - ? ? ? ? ? ? ? -

?? ?3.修改配置文件
vim /oldboyedu/softwares/kafka_2.13-3.2.1/config/consumer.properties?
...
group.id=linux82-elk
?
?? ?
?? ?4.基于配置文件啟動(dòng)消費(fèi)者
kafka-console-consumer.sh --bootstrap-server 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 --topic oldboyedu-linux82 --from-beginning --consumer.config /oldboyedu/softwares/kafka_2.13-3.2.1/config/consumer.properties

?? ?5.再次查看消費(fèi)者信息
kafka-consumer-groups.sh --bootstrap-server 10.0.0.101:9092 ?--describe --group linux82-elk

kafka生產(chǎn)環(huán)境應(yīng)該監(jiān)控哪些指標(biāo):
?? ?(1)監(jiān)控kafka的程序是否正常運(yùn)行;
?? ?(2)數(shù)據(jù)是否有延遲;
?? ?(3)是否頻繁涉及到ISR列表的變化;
?? ?(4)iops;
?? ?(5)其他參數(shù)可參考<<kafka權(quán)威指南>>;

安裝開源工具監(jiān)控kafka集群:
?? ?1.安裝MySQL數(shù)據(jù)庫并授權(quán)
?? ??? ?1.1 部署mariadb
yum -y install mariadb-server
systemctl start mariadb

?? ??? ?1.2 修改配置文件
vim /etc/my.cnf
...
[mysqld]
# 跳過名稱解析。
skip-name-resolve

?? ??? ?1.3 重啟服務(wù)并測試鏈接
systemctl restart mariadb
mysql -u linux82 -poldboyedu -h 10.0.0.103

?? ??? ?1.4 創(chuàng)建數(shù)據(jù)庫
CREATE DATABASE oldboyedu_linux82 DEFAULT CHARACTER SET utf8mb4;
?? ??? ?
?? ??? ?1.5 授權(quán)
CREATE USER linux82 IDENTIFIED BY 'oldboyedu';
GRANT ALL ON oldboyedu_linux82.* TO linux82;
SHOW GRANTS FOR linux82;

?? ??? ?
?? ??? ?
?? ?2.修改kafka開啟JMX
?? ??? ?2.1 修改啟動(dòng)腳本
vim /oldboyedu/softwares/kafka_2.13-3.2.1/bin/kafka-server-start.sh

...
# 大概在30行左右
?? ?export KAFKA_HEAP_OPTS="-Xmx256m -Xms256m"
? ? export JMX_PORT="8888"

?? ??? ?2.2 同步啟動(dòng)腳本到其他節(jié)點(diǎn)
data_rsync.sh /oldboyedu/softwares/kafka_2.13-3.2.1/bin/kafka-server-start.sh

?? ??? ?2.3 所有重啟kafka服務(wù)
kafka-server-stop.sh?
kafka-server-start.sh $KAFKA_HOME/config/server.properties &>/dev/null &
?? ?
?? ?

?? ?3.啟動(dòng)zookeeper的JMX端口
?? ??? ?3.1 修改啟動(dòng)腳本
vim /oldboyedu/softwares/apache-zookeeper-3.8.0-bin/bin/zkServer.sh +77
...
# 大概是在77行左右
ZOOMAIN="-Dzookeeper.4lw.commands.whitelist=* ${ZOOMAIN}"

?? ?
?? ??? ?3.2 重啟zookeeper服務(wù)
zkServer.sh restart

?? ?4.安裝kafka eagle監(jiān)控工具
?? ??? ?4.1 解壓軟件包
unzip kafka-eagle-bin-2.0.8.zip
tar xf efak-web-2.0.8-bin.tar.gz ?-C /oldboyedu/softwares/

?? ??? ?4.2 修改kafka eagle啟動(dòng)腳本的堆內(nèi)存大小
vim ?/oldboyedu/softwares/efak-web-2.0.8/bin/ke.sh?
...
export KE_JAVA_OPTS="-server -Xmx256m -Xms256m -XX:MaxGCPauseMillis=20 -XX:+UseG1GC -XX:MetaspaceSize=128m -XX:InitiatingHeapOccupancyPercent=35 -XX:G1HeapRegionSize=16M -XX:MinMetaspaceFreeRatio=50 -XX:MaxMetaspaceFreeRatio=80"

?? ??? ?4.3 修改配置文件?
cat > /oldboyedu/softwares/efak-web-2.0.8/conf/system-config.properties <<EOF
efak.zk.cluster.alias=linux82,cluster2
linux82.zk.list=10.0.0.101:2181/oldboyedu-linux82-kafka3.2.1
cluster2.zk.list=10.0.0.101:2181/oldboyedu-linux82-kafka080
linux82.efak.broker.size=20
kafka.zk.limit.size=32
efak.webui.port=8048
linux82.efak.offset.storage=kafka
cluster2.efak.offset.storage=zk
linux82.efak.jmx.uri=service:jmx:rmi:///jndi/rmi://%s/jmxrmi
efak.metrics.charts=true
efak.metrics.retain=15
efak.sql.topic.records.max=5000
efak.sql.topic.preview.records.max=10
efak.topic.token=keadmin
efak.driver=com.mysql.cj.jdbc.Driver
efak.url=jdbc:mysql://10.0.0.103:3306/oldboyedu_linux82?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull
efak.username=linux82
efak.password=oldboyedu
EOF

?? ??? ?4.4 配置環(huán)境變量
cat > /etc/profile.d/kafka_eagle.sh <<EOF
#!/bin/bash

export KE_HOME=/oldboyedu/softwares/efak-web-2.0.8
export PATH=$PATH:$KE_HOME/bin
EOF

?? ??? ?4.5 啟動(dòng)kafka eagle服務(wù)，第一次需要等一會(huì)，可能有點(diǎn)漫長哈！大概30s~1分鐘左右
ke.sh start

kafka集群壓力測試:
?? ?1.簡介?
所謂壓力測試就是對(duì)一個(gè)集群的處理能力的上限做一個(gè)評(píng)估。為將來集群擴(kuò)容提供有效的依據(jù)。
?? ?
?? ?2.為什么要進(jìn)行壓力測試
(1)壓力測試可以了解當(dāng)前集群的處理能力上限;
(2)當(dāng)修改集群的配置參數(shù)后，壓力測試可以協(xié)助運(yùn)維人員去參考本次調(diào)優(yōu)的效果;
(3)壓力測試的結(jié)果以后可以作為參考擴(kuò)容集群的有效依據(jù);
?? ?
?? ?3.實(shí)戰(zhàn)案例
install -d /tmp/kafka-test/
cat > oldboyedu-kafka-test.sh <<'EOF'
# 創(chuàng)建topic
kafka-topics.sh --bootstrap-server 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 --topic oldboyedu-kafka-2022 --replication-factor 1 --partitions 10 --create

# 啟動(dòng)消費(fèi)者消費(fèi)數(shù)據(jù)
nohup kafka-consumer-perf-test.sh --broker-list 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 --topic ?oldboyedu-kafka-2022 --messages 100000000 &>/tmp/kafka-test/oldboyedu-kafka-consumer.log &

# 啟動(dòng)生產(chǎn)者寫入數(shù)據(jù)
nohup kafka-producer-perf-test.sh --num-records 100000000 --record-size 1000 --topic ?oldboyedu-kafka-2022 --throughput 1000000 --producer-props bootstrap.servers=10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 &> /tmp/kafka-test/oldboyedu-kafka-producer.log &
EOF
bash oldboyedu-kafka-test.sh?

參數(shù)說明:
kafka-consumer-perf-test.sh?
?? ?---messages:
? ? ?? ?指定消費(fèi)消息的數(shù)量。
?? ?--broker-list:
? ? ?? ?指定broker列表。
? ? --topic:
? ? ?? ?指定topic主體。
? ? ?? ?
kafka-producer-perf-test.sh?
? ? -num-records
? ? ? ? 生產(chǎn)消息的數(shù)量。
? ? --record-size:
? ? ? ? 每條消息的大小，單位是字節(jié)。
? ? --topic:
? ? ?? ?指定topic主體。
?? ?--throughput?
?? ??? ?設(shè)置每秒發(fā)送的消息數(shù)量，即指定最大消息的吞吐量，若設(shè)置為-1代表不限制！
?? ?--producer-props bootstrap.servers
?? ??? ?指定broker列表。
?? ??? ?
?? ??? ?
溫馨提示:
?? ?本案例測試大約會(huì)生成93GB( echo ?"100000000 * 1000/1024/1024/1024" | bc)的數(shù)據(jù)，如果硬盤資源不足的小伙伴可以暫時(shí)不用測試了，或者該小上面提到的參數(shù)。

今日內(nèi)容回顧:
?? ?- kafka的ISR工作機(jī)制原理 ? *****
?? ?- ZK的znode基礎(chǔ)操作 ?***
?? ?- kafka多實(shí)例部署
?? ?- kafka eagle監(jiān)控工具
?? ?- kafka集群壓力測試 ?***

下次課程:
?? ?- kafka優(yōu)化
?? ?- kafka配合elk使用案例
? ? - zk的ACL
?? ?- zk的監(jiān)控工具
?? ?- zk調(diào)優(yōu)
?? ?- ES8集群環(huán)境搭建文章來源地址http://www.zghlxwxcb.cn/news/detail-691046.html

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