柚子快報邀請碼778899分享：大數(shù)據(jù)-kafka學(xué)習(xí)筆記

http://yzkb.51969.com/

先自我介紹一下，小編浙江大學(xué)畢業(yè)，去過華為、字節(jié)跳動等大廠，目前阿里P7

深知大多數(shù)程序員，想要提升技能，往往是自己摸索成長，但自己不成體系的自學(xué)效果低效又漫長，而且極易碰到天花板技術(shù)停滯不前！

因此收集整理了一份《2024年最新大數(shù)據(jù)全套學(xué)習(xí)資料》，初衷也很簡單，就是希望能夠幫助到想自學(xué)提升又不知道該從何學(xué)起的朋友。

既有適合小白學(xué)習(xí)的零基礎(chǔ)資料，也有適合3年以上經(jīng)驗的小伙伴深入學(xué)習(xí)提升的進階課程，涵蓋了95%以上大數(shù)據(jù)知識點，真正體系化！

由于文件比較多，這里只是將部分目錄截圖出來，全套包含大廠面經(jīng)、學(xué)習(xí)筆記、源碼講義、實戰(zhàn)項目、大綱路線、講解視頻，并且后續(xù)會持續(xù)更新

如果你需要這些資料，可以添加V獲?。簐ip204888 （備注大數(shù)據(jù)）

正文

消費者消費一個分區(qū)中的數(shù)據(jù)時，會跟蹤他們自己消費到的偏移量，Kafka 會定期將偏移量提交到 Kafka 主題中的特殊主題（__consumer_offsets）中，這樣，消費者如果停止或重新啟動后，會從上次的偏移量繼續(xù)消費。偏移量是每條消息在分區(qū)中的位置。

消費者組

由多個kafka消費者組成的一組消費者組，用于同時消費處理kafka中一個主題所有分區(qū)中的數(shù)據(jù)。只要在創(chuàng)建kafka消費者的時候?qū)roup id設(shè)置成一樣的，那么就可以創(chuàng)建多個消費者構(gòu)成消費者組了。

一個主題的一個分區(qū)只能由一個消費者組內(nèi)的一個消費者處理，否則會導(dǎo)致數(shù)據(jù)重復(fù)消費。一個消費者組的每個消費者負(fù)責(zé)消費不同分區(qū)的數(shù)據(jù)。

消費者組的好處：加快消費處理數(shù)據(jù)的速度，橫向提高整個消費能力。如下圖，一開始就一個消費者c1，他要自己一個人消費處理來自topicA主題的四個分區(qū)的數(shù)據(jù)，而我們可以增加三個消費者c2、c3、c4和c1構(gòu)成一個消費者組來同時消費處理topicA主題的四個分區(qū)的數(shù)據(jù)，這樣消費處理數(shù)據(jù)的速度就提升了。（前提是有多個分區(qū)）

主題

上面那樣肯定不好，各種消息的的生產(chǎn)者（生產(chǎn)圓蛋蛋、生產(chǎn)方框框、生產(chǎn)小心心）將消息都發(fā)給kafka，然后kafka將消息都分類，每種分類都有相應(yīng)的主題，然后消費者根據(jù)需要訂閱相應(yīng)的主題。就能收到對應(yīng)的消息。

分區(qū)

如果一個主題的消息比較多，就可以考慮分區(qū)，分區(qū)可以分布在不同的服務(wù)器上，所以主題也可以分布在不同的服務(wù)器上，這樣比單服務(wù)器處理快。

如果生成者沒有指定分區(qū)，分區(qū)器就會根據(jù)每條消息的鍵算出消息該去哪個分區(qū)。鍵：就是每條消息的一個標(biāo)記，決定了消息該去哪個分區(qū)。分區(qū)器：就是一個算法，算消息該去哪個分區(qū)，輸入是鍵，輸出是消息去的分區(qū)。

偏移量

偏移量就是消息在每個分區(qū)中的位置，kafka在收到消息的時候，會為每個消息設(shè)置偏移量，然后將消息存到磁盤中。

消費者只能按順序消費讀取。消費者如果要分區(qū)0的第四個，kafka就會說第三個還沒讀取，不給第四個。

kafka集群

一個broker就是一個kafka服務(wù)器。下面有兩個broker構(gòu)成了kafka集群，他們的數(shù)據(jù)通過復(fù)制同步，當(dāng)有一個kafka宕機了，另一臺就可以先頂上，保證了kafka的可靠性。

監(jiān)控kafka

這個前提得先安裝jdk

1、修改kafka的啟動腳本

vim bin/kafka-server-start.sh

if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then

export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"

fi

改為

if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then

export KAFKA_HEAP_OPTS="-server -Xms2G -Xmx2G

-XX:PermSize=128m -XX:+UseG1GC -XX:MaxGCPauseMillis=200

-XX:ParallelGCThreads=8 -XX:ConcGCThreads=5

-XX:InitiatingHeapOccupancyPercent=70"

export JMX_PORT="9999"

#export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"

fi

修改kafka進程信息：

-Xms2G：設(shè)置 Kafka 進程的初始堆內(nèi)存大小為 2 GB。

-Xmx2G：設(shè)置 Kafka 進程的最大堆內(nèi)存大小為 2 GB。

XX:PermSize=128m：設(shè)置持久代（PermGen）的初始大小為 128 MB。請注意，這個選項在 Java 8 和更新的版本中不再適用，因為 PermGen 已被 Metaspace 取代。

-XX:+UseG1GC：指定使用 G1 垃圾收集器。

-XX:MaxGCPauseMillis=200：設(shè)置最大垃圾收集暫停時間為 200 毫秒。

XX:ParallelGCThreads=8：設(shè)置并行垃圾收集線程的數(shù)量為 8。

XX:ConcGCThreads=5：設(shè)置并發(fā)垃圾收集線程的數(shù)量為 5。

XX:InitiatingHeapOccupancyPercent=70：設(shè)置堆內(nèi)存占用百分比，當(dāng)堆內(nèi)存使用達到 70% 時，啟動并發(fā)垃圾收集。

這些參數(shù)的目的是調(diào)整 Kafka 進程的性能和垃圾收集行為，以滿足特定的性能需求。請注意，這些參數(shù)的值可以根據(jù)你的 Kafka 部署和硬件資源進行調(diào)整。堆內(nèi)存的大小和垃圾收集器的選擇將影響 Kafka 的性能和穩(wěn)定性。

最后，這段腳本還設(shè)置了 JMX 端口為 9999，這是用于監(jiān)控 Kafka 進程的 Java Management Extensions（JMX）端口。通過此端口，你可以使用 JMX 工具監(jiān)控 Kafka 進程的性能指標(biāo)和狀態(tài)。如果需要監(jiān)控 Kafka，你可以使用 JMX 工具連接到此端口。

2、官網(wǎng)下載安裝包

https://www.kafka-eagle.org/

3、上傳解壓

第一次解壓后，里面有個壓縮包再解壓才是真正的。

/opt/module/efak/conf/system-config.properties

5、配置環(huán)境變量

$ sudo vim /etc/profile.d/my_env.sh

# kafkaEFAK

export KE_HOME=/opt/module/efak

export PATH=$PATH:$KE_HOME/bin

source /etc/profile

6、啟動

/bin/kf.sh start

壓力測試

# 單Kafka服務(wù)器，生成者發(fā)送1000000條數(shù)據(jù)，每條大小1k，總共發(fā)送大約

bin/kafka-producer-perf-test.sh --topic test --record-size 1024 --num-records 1000000 --throughput 10000 --producer-props bootstrap.servers=linjl:9092 batch.size=16384 linger.ms=0

batch.size=16384 linger.ms=0 9.76 MB/sec

record-size 是一條信息有多大，單位是字節(jié)，本次測試設(shè)置為 1k。

BUG

1、Error while fetching metadata with correlation id : {LEADER_NOT_AVAILABLE}

2、

[root@linjl kafka_2.12-3.0.0]# ./bin/kafka-console-consumer.sh --topic quickstart-events --bootstrap-server linjl:9092

[2023-09-13 16:51:54,710] WARN [Consumer clientId=consumer-console-consumer-32025-1, groupId=console-consumer-32025] Error while fetching metadata with correlation id 2 : {quickstart-events=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)

這個警告消息 “Error while fetching metadata with correlation id 2 : {quickstart-events=LEADER_NOT_AVAILABLE}” 表示 Kafka 消費者在嘗試訂閱主題 “quickstart-events” 時遇到了 “LEADER_NOT_AVAILABLE” 錯誤。這個錯誤通常表示消費者無法找到主題的 leader 分區(qū)，因此它無法讀取消息。

我的猜想： 可能是因為 Kafka 服務(wù)器無法從 ZooKeeper 獲取到有關(guān) “quickstart-events” 主題的元數(shù)據(jù)信息，包括分區(qū)的 Leader 信息。

3、Received invalid metadata error in produce request on partition quickstart-events-0

due to org.apache.kafka.common.errors.KafkaStorageException: Disk error when trying to access log file on the disk… Going to request metadata update now (org.apache.kafka.clients.producer.internals.Sender)

表示在嘗試將消息寫入分區(qū) “quickstart-events-0” 時，Kafka 生產(chǎn)者遇到了磁盤錯誤，無法訪問日志文件。這個錯誤通常與磁盤故障或磁盤空間不足有關(guān)。

4、Java客戶端創(chuàng)建生產(chǎn)者，發(fā)送消息給kafka沒響應(yīng)。

Properties properties = new Properties();

properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.239.128:9092");

// key,value 序列化（必須）：key.serializer，value.serializer

properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName())

網(wǎng)絡(luò)連接都能通，而且防火墻也都關(guān)了。

解決：在server.properties配置文件中配置

# The address the socket server listens on. If not configured, the host name will be equal to the value of

# java.net.InetAddress.getCanonicalHostName(), with PLAINTEXT listener name, and port 9092.

# FORMAT:

# listeners = listener_name://host_name:port

# EXAMPLE:

# listeners = PLAINTEXT://your.host.name:9092

listeners=PLAINTEXT://192.168.239.128:9092

kafka和flink結(jié)合案例

數(shù)據(jù)寫入kafka，flink訂閱消費

安裝kafka單服務(wù)

1、官方下載地址：http://kafka.apache.org/downloads.html

2、解壓安裝包

下載完將安裝包上傳到centos中，然后解壓

$ tar -zxvf kafka_2.12-3.0.0.tgz -C /opt/module/

3、 修改解壓后的文件名稱

$ mv kafka_2.12-3.0.0/ kafka

4、進入到/opt/module/kafka 目錄，修改配置文件

$ cd config/

$ vim server.properties

#broker 的全局唯一編號，不能重復(fù)，只能是數(shù)字。

broker.id=0

#處理網(wǎng)絡(luò)請求的線程數(shù)量

num.network.threads=3

#用來處理磁盤 IO 的線程數(shù)量

num.io.threads=8

#發(fā)送套接字的緩沖區(qū)大小

socket.send.buffer.bytes=102400

#接收套接字的緩沖區(qū)大小

socket.receive.buffer.bytes=102400

#請求套接字的緩沖區(qū)大小

socket.request.max.bytes=104857600

#kafka 運行日志(數(shù)據(jù))存放的路徑，路徑不需要提前創(chuàng)建，kafka 自動幫你創(chuàng)建，可以配置多個磁盤路徑，路徑與路徑之間可以用"，"分隔

log.dirs=/opt/module/kafka/datas

#topic 在當(dāng)前 broker 上的分區(qū)個數(shù)

num.partitions=1

#用來恢復(fù)和清理 data 下數(shù)據(jù)的線程數(shù)量

num.recovery.threads.per.data.dir=1

# 每個 topic 創(chuàng)建時的副本數(shù)，默認(rèn)時 1 個副本

offsets.topic.replication.factor=1

#segment 文件保留的最長時間，超時將被刪除

log.retention.hours=168

#每個 segment 文件的大小，默認(rèn)最大 1G

log.segment.bytes=1073741824

# 檢查過期數(shù)據(jù)的時間，默認(rèn) 5 分鐘檢查一次是否數(shù)據(jù)過期

log.retention.check.interval.ms=300000

#配置連接Zookeeper 集群地址（在 zk 根目錄下創(chuàng)建/kafka，方便管理）

zookeeper.connect=hadoop102:2181,hadoop103:2181,hadoop104:2181/kafka

5、配置kafka環(huán)境變量

vim /etc/profile.d/my_env.sh

**網(wǎng)上學(xué)習(xí)資料一大堆，但如果學(xué)到的知識不成體系，遇到問題時只是淺嘗輒止，不再深入研究，那么很難做到真正的技術(shù)提升。**

**需要這份系統(tǒng)化的資料的朋友，可以添加V獲?。簐ip204888 （備注大數(shù)據(jù)）**


**一個人可以走的很快，但一群人才能走的更遠！不論你是正從事IT行業(yè)的老鳥或是對IT行業(yè)感興趣的新人，都歡迎加入我們的的圈子（技術(shù)交流、學(xué)習(xí)資源、職場吐槽、大廠內(nèi)推、面試輔導(dǎo)），讓我們一起學(xué)習(xí)成長！**

2181,hadoop103:2181,hadoop104:2181/kafka

5、配置kafka環(huán)境變量

vim /etc/profile.d/my_env.sh

**網(wǎng)上學(xué)習(xí)資料一大堆，但如果學(xué)到的知識不成體系，遇到問題時只是淺嘗輒止，不再深入研究，那么很難做到真正的技術(shù)提升。**

**需要這份系統(tǒng)化的資料的朋友，可以添加V獲?。簐ip204888 （備注大數(shù)據(jù)）**

[外鏈圖片轉(zhuǎn)存中...(img-iJxxV7Ho-1713181446355)]

**一個人可以走的很快，但一群人才能走的更遠！不論你是正從事IT行業(yè)的老鳥或是對IT行業(yè)感興趣的新人，都歡迎加入我們的的圈子（技術(shù)交流、學(xué)習(xí)資源、職場吐槽、大廠內(nèi)推、面試輔導(dǎo)），讓我們一起學(xué)習(xí)成長！**

柚子快報邀請碼778899分享：大數(shù)據(jù)-kafka學(xué)習(xí)筆記

http://yzkb.51969.com/

相關(guān)鏈接