大數據技術之_10_Kafka學習_Kafka概述+Kafka集羣部署+Kafka工做流程分析+Kafka API實戰+Kafka Producer攔截器+Kafka Streams
第1章 Kafka概述1.1 消息隊列1.2 爲何須要消息隊列1.3 什麼是Kafka1.4 Kafka架構第2章 Kafka集羣部署2.1 環境準備2.1.1 集羣規劃2.1.2 jar包下載2.2 Kafka集羣部署2.3 Kafka命令行操做第3章 Kafka工做流程分析3.1 Kafka 生產過程分析3.1.1 寫入方式3.1.2 分區（Partition）3.1.3 副本（Replication）3.1.4 寫入流程3.2 Broker 保存消息3.2.1 存儲方式3.2.2 存儲策略3.2.3 Zookeeper存儲結構3.3 Kafka 消費過程分析3.3.1 高級API3.3.2 低級API3.3.3 消費者組3.3.4 消費方式3.3.5 消費者組案例第4章 Kafka API實戰4.1 環境準備4.2 Kafka生產者Java API4.2.1 建立生產者（過期的API）4.2.2 建立生產者（新的API）4.2.3 建立生產者帶回調函數（新的API）4.2.4 自定義分區生產者4.3 Kafka消費者Java API4.3.1 高級API4.3.2 低級API第5章 Kafka Producer攔截器(interceptor)5.1 攔截器原理5.2 攔截器案例第6章 Kafka Streams6.1 概述6.1.1 Kafka Streams6.1.2 Kafka Streams 特色6.1.3 爲何要有 Kafka Stream？6.2 Kafka Stream 數據清洗案例第7章 擴展知識7.1 Kafka 與 Flume 比較7.2 Flume 與 kafka 集成7.3 Kafka配置信息7.3.1 Broker 配置信息7.3.2 Producer 配置信息7.3.3 Consumer 配置信息7.4 如何查看 Kafka 集羣維護的 offset 信息javascript
第1章 Kafka概述
1.1 消息隊列
1）點對點模式（一對一，消費者主動拉取數據，消息收到後消息清除）
點對點模型一般是一個基於拉取或者輪詢的消息傳送模型，這種模型從隊列中請求信息，而不是將消息推送到客戶端。這個模型的特色是發送到隊列的消息被一個且只有一個接收者接收處理，即便有多個消息監聽者也是如此。php
2）發佈/訂閱模式（一對多，數據生產後，推送給全部訂閱者）
發佈訂閱模型則是一個基於推送的消息傳送模型。發佈訂閱模型能夠有多種不一樣的訂閱者，臨時訂閱者只在主動監聽主題時才接收消息，而持久訂閱者則監聽主題的全部消息，即便當前訂閱者不可用，處於離線狀態。html
1.2 爲何須要消息隊列
1）解耦：
容許你獨立的擴展或修改兩邊的處理過程，只要確保它們遵照一樣的接口約束。java
2）冗餘：
消息隊列把數據進行持久化直到它們已經被徹底處理，經過這一方式規避了數據丟失風險。許多消息隊列所採用的"插入-獲取-刪除"範式中，在把一個消息從隊列中刪除以前，須要你的處理系統明確的指出該消息已經被處理完畢，從而確保你的數據被安全的保存直到你使用完畢。node
3）擴展性：
由於消息隊列解耦了你的處理過程，因此增大消息入隊和處理的頻率是很容易的，只要另外增長處理過程便可。nginx
4）靈活性 & 峯值處理能力：
在訪問量劇增的狀況下，應用仍然須要繼續發揮做用，可是這樣的突發流量並不常見。若是爲以能處理這類峯值訪問爲標準來投入資源隨時待命無疑是巨大的浪費。使用消息隊列可以使關鍵組件頂住突發的訪問壓力，而不會由於突發的超負荷的請求而徹底崩潰。web
5）可恢復性：
系統的一部分組件失效時，不會影響到整個系統。消息隊列下降了進程間的耦合度，因此即便一個處理消息的進程掛掉，加入隊列中的消息仍然能夠在系統恢復後被處理。sql
6）順序保證：
在大多使用場景下，數據處理的順序都很重要。大部分消息隊列原本就是排序的，而且能保證數據會按照特定的順序來處理。（Kafka保證一個Partition內的消息的有序性）shell
7）緩衝：
有助於控制和優化數據流通過系統的速度，解決生產消息和消費消息的處理速度不一致的狀況。apache
8）異步通訊：
不少時候，用戶不想也不須要當即處理消息。消息隊列提供了異步處理機制，容許用戶把一個消息放入隊列，但並不當即處理它。想向隊列中放入多少消息就放多少，而後在須要的時候再去處理它們。
1.3 什麼是Kafka
Kafka 是最初由 Linkedin 公司開發，是一個分佈式、支持分區的（partition）、多副本的（replica），基於 zookeeper 協調的分佈式消息系統，它的最大的特性就是能夠實時的處理大量數據以知足各類需求場景：好比基於hadoop的批處理系統、低延遲的實時系統、storm/Spark流式處理引擎，web/nginx日誌、訪問日誌，消息服務等等，用scala語言編寫，Linkedin 於 2010 年貢獻給了 Apache 基金會併成爲頂級開源項目。
在流式計算中，Kafka 通常用來緩存數據，Storm經過消費Kafka的數據進行計算。
Kafka 是基於點對點模式的消息隊列。
1）Apache Kafka是一個開源消息系統，由 Scala 寫成。是由 Apache 軟件基金會開發的一個開源消息系統項目。
2）Kafka 最初是由LinkedIn公司開發，並於 2011 年初開源。2012 年 10月從Apache Incubator畢業，併成爲頂級開源項目。該項目的目標是爲處理實時數據提供一個統1、高通量、低等待的平臺。
3）Kafka 是一個分佈式消息隊列。Kafka 對消息保存時根據 Topic 進行歸類，發送消息者稱爲 Producer，消息接受者稱爲Consumer，此外 kafka 集羣有多個 kafka 實例組成，每一個實例(server)稱爲 broker。真正存儲數據的地方叫作 Topic。
4）不管是 kafka 集羣，仍是 Consumer 都依賴於Zookeeper集羣保存一些meta信息，來保證系統可用性。
1.4 Kafka架構
Kafka總體架構圖
Kafka總體架構圖詳解
1）Producer ：消息生產者，就是向 kafka broker 發消息的客戶端。
2）Consumer ：消息消費者，向 kafka broker 取消息的客戶端。
3）Topic ：能夠理解爲一個隊列。
4） Consumer Group（CG）：這是 kafka 用來實現一個 topic 消息的廣播（發給全部的consumer）和單播（發給任意一個consumer）的手段。一個 topic 能夠有多個 CG。topic 的消息會複製（不是真的複製，是概念上的）到全部的 CG，但每一個 partion 只會把消息發給該CG中的一個 consumer。若是須要實現廣播，只要每 consumer 有一個獨立的 CG 就能夠了。要實現單播只要全部的 consumer 在同一個CG。用CG還能夠將 consumer 進行自由的分組而不須要屢次發送消息到不一樣的 topic。
5）Broker ：一臺 kafka 服務器就是一個 broker。一個集羣由多個 broker 組成。一 個broker 能夠容納多個 topic。
6）Partition：爲了實現擴展性，一個很是大的 topic 能夠分佈到多個 broker（即服務器）上，一個 topic 能夠分爲多個 partition，每一個 partition 是一個有序的隊列。partition 中的每條消息都會被分配一個有序的id（offset）。kafka只保證按一個 partition 中的順序將消息發給consumer，不保證一個 topic 的總體（多個partition間）的順序。
7）Offset：kafka 的存儲文件都是按照 offset.kafka 來命名，用 offset 作名字的好處是方便查找。例如你想找位於2049的位置，只要找到 2048.kafka 的文件便可。固然 the first offset 就是 00000000000.kafka。
8）分區對於Kafka集羣的好處是：實現負載均衡。分區對於消費者來講，能夠提升併發度，提升效率。在公司中應用的時候，針對於某一個 Topic，它有幾個分區(n個)，咱們就對應的建一個有幾個消費者的消費者組(m個)。即：n大於或者等於m，最好是n=m。當n>m時，就意味着某一個消費者會消費多個分區的數據。不只如此，一個消費者還能夠消費多個 Topic 數據。
第2章 Kafka集羣部署
2.1 環境準備
2.1.1 集羣規劃
hadoop102 hadoop103 hadoop104
zk zk zk
kafka kafka kafka
2.1.2 jar包下載
http://kafka.apache.org/downloads.html
2.2 Kafka集羣部署
1）解壓安裝包
[atguigu@hadoop102 software]$ tar -zxvf kafka_2.11-0.11.0.2.tgz -C /opt/module/
2）修改解壓後的文件名稱
[atguigu@hadoop102 module]$ mv kafka_2.11-0.11.0.2/ kafka
3）在/opt/module/kafka目錄下建立logs文件夾
[atguigu@hadoop102 kafka]$ mkdir logs
4）修改配置文件
[atguigu@hadoop102 kafka]$ cd config/
[atguigu@hadoop102 config]$ vim server.properties
輸入如下內容：
#broker的全局惟一編號，不能重複
broker.id=0
#刪除topic功能使能
delete.topic.enable=true
#處理網絡請求的線程數量
num.network.threads=3
#用來處理磁盤IO的現成數量
num.io.threads=8
#發送套接字的緩衝區大小
socket.send.buffer.bytes=102400
#接收套接字的緩衝區大小
socket.receive.buffer.bytes=102400
#請求套接字的緩衝區大小
socket.request.max.bytes=104857600
#kafka運行日誌存放的路徑
log.dirs=/opt/module/kafka/logs
#topic在當前broker上的分區個數
num.partitions=1
#用來恢復和清理data下數據的線程數量
num.recovery.threads.per.data.dir=1
#segment文件保留的最長時間，超時將被刪除（單位小時）
log.retention.hours=168
#配置鏈接Zookeeper集羣地址
zookeeper.connect=hadoop102:2181,hadoop103:2181,hadoop104:2181
5）配置環境變量
[atguigu@hadoop102 module]$ sudo vim /etc/profile
#KAFKA_HOME
export KAFKA_HOME=/opt/module/kafka
export PATH=$PATH:$KAFKA_HOME/bin
[atguigu@hadoop102 module]$ source /etc/profile
6）分發安裝包
[atguigu@hadoop102 module]$ xsync kafka/
注意：分發以後記得配置其餘機器的環境變量。
7）分別在hadoop103和hadoop104上修改配置文件/opt/module/kafka/config/server.properties中的broker.id=一、broker.id=2注：broker.id不得重複。
8）啓動Kafka集羣
依次在hadoop10二、hadoop10三、hadoop104節點上啓動kafka
[atguigu@hadoop102 kafka]$ bin/kafka-server-start.sh config/server.properties &
[atguigu@hadoop103 kafka]$ bin/kafka-server-start.sh config/server.properties &
[atguigu@hadoop104 kafka]$ bin/kafka-server-start.sh config/server.properties &
啓動Kafka是一個阻塞進程，會打印咱們操做kafka的日誌，咱們能夠把窗口放到後臺，在命令後面加一個與&符號，將該阻塞進程放到後臺。
寫羣起Kafka集羣腳本的時候，須要使用-daemon命令，具體以下：
[atguigu@hadoop102 kafka]$ bin/kafka-server-start.sh -daemon config/server.properties
[atguigu@hadoop103 kafka]$ bin/kafka-server-start.sh -daemon config/server.properties
[atguigu@hadoop104 kafka]$ bin/kafka-server-start.sh -daemon config/server.properties
-daemon 表示守護進程，會將日誌打印在後臺。
9）關閉Kafka集羣
[atguigu@hadoop102 kafka]$ bin/kafka-server-stop.sh stop
[atguigu@hadoop103 kafka]$ bin/kafka-server-stop.sh stop
[atguigu@hadoop104 kafka]$ bin/kafka-server-stop.sh stop
寫羣起Kafka集羣腳本的時候，須要使用-daemon命令，具體以下：
[atguigu@hadoop102 kafka]$ bin/kafka-server-stop.sh -daemon
[atguigu@hadoop103 kafka]$ bin/kafka-server-stop.sh -daemon
[atguigu@hadoop104 kafka]$ bin/kafka-server-stop.sh -daemon
2.3 Kafka命令行操做
0）補充知識
jps 查看當前進程
jps -l 查看當前進程所屬主類
`注意：`當有不少進程都是同一個名字，咱們該如何區分？
`答：`每一次啓動一個進程後，咱們將該進程與對應的進程ID寫入一個文檔中。若是某一個進程出現問題或者某一個框架出現問題，便於咱們kill掉相應的進程。不至於關閉整個系統。（生產環境下通常不容許關閉或重啓整個系統！）
1）查看當前服務器中的全部topic
[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --zookeeper hadoop102:2181 --list
2）建立topic
[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --zookeeper hadoop102:2181 \
--create --replication-factor 3 --partitions 1 --topic first
bin/kafka-topics.sh --zookeeper hadoop102:2181 \
--create --replication-factor 2 --partitions 3 --topic second
選項說明：
--topic 定義topic名
--replication-factor 定義副本數（注：副本數不能大於節點數，不然會報錯！）
--partitions 定義分區數
3）刪除topic
[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --zookeeper hadoop102:2181 \
--delete --topic first
注意：須要server.properties中設置delete.topic.enable=true不然只是標記刪除或者直接重啓。
4）發送消息（生產者鏈接的是kafka集羣默認的端口號是：9092）
[atguigu@hadoop102 kafka]$ bin/kafka-console-producer.sh \
--broker-list hadoop102:9092 --topic first
>hello world
>atguigu atguigu
`注意：`生產者鏈接的是kafka集羣。
5）消費消息
老版本
[atguigu@hadoop103 kafka]$ bin/kafka-console-consumer.sh \
--zookeeper hadoop102:2181 --from-beginning --topic first
或者
新版本
[atguigu@hadoop103 kafka]$ bin/kafka-console-consumer.sh \
--bootstrap-server hadoop102:9092 --from-beginning --topic first
`注意：`消費者會將本身的 offset 文件保存在 zookeeper(低版本的 kafka)。因此消費者鏈接的是 zookeeper。
--from-beginning：會把first主題中以往全部的數據都讀取出來。根據業務場景選擇是否增長該配置。若是不加該配置，那麼消費者消費的消息將是最新的消息(不包括以往的全部數據)。
6）查看某個topic的詳情
[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --zookeeper hadoop102:2181 \
--describe --topic first
Topic:first PartitionCount:1 ReplicationFactor:3 Configs:
Topic: first Partition: 0 Leader: 2 Replicas: 2,0,1 Isr: 2,0,1
Isr的做用：當 leader 掛掉後，選舉新 leader 時使用的。Isr 的排序規則是：與 leader 的類似度，越高越在前，越在前越有可能成爲新 leader。
7）警告問題解釋
[atguigu@hadoop103 kafka]$ bin/kafka-console-consumer.sh \
> --zookeeper hadoop102:2181 --from-beginning --topic first
Using the ConsoleConsumer with old consumer is deprecated and will be removed in a future major release. Consider using the new consumer by passing [bootstrap-server] instead of [zookeeper].
在高版本的kafka中，消費者會將本身的 offset文件 保存在 kafka 集羣的本地，不交給 zookeeper 維護了！以下圖所示：
這樣作的好處是：提升了效率，減小了網絡傳輸。
第3章 Kafka工做流程分析
3.1 Kafka 生產過程分析
3.1.1 寫入方式
producer 採用推（push）模式將消息發佈到 broker，每條消息都被追加（append）到分區（patition）中，屬於順序寫磁盤（順序寫磁盤效率比隨機寫內存要高，保障kafka吞吐率）。
3.1.2 分區（Partition）
消息發送時都被髮送到一個 topic，其本質就是一個目錄，而topic是由一些 Partition Logs(分區日誌)組成，其組織結構以下圖所示：
咱們能夠看到，每一個 Partition 中的消息都是有序的，生產的消息被不斷追加到 Partition log 上，其中的每個消息都被賦予了一個惟一的 offset值。
1）分區的緣由
（1）方便在集羣中擴展，每一個 Partition 能夠經過調整以適應它所在的機器，而一個topic又能夠有多個 Partition 組成，所以整個集羣就能夠適應任意大小的數據了。
（2）能夠提升併發，由於能夠以 Partition 爲單位讀寫了。
2）分區的原則
（1）指定了 patition，則直接使用。
（2）未指定 patition 但指定 key，經過對 key 的 value 進行 hash 出一個 patition。
（3）patition 和 key 都未指定，使用輪詢選出一個 patition。
DefaultPartitioner類
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
int numPartitions = partitions.size();
if (keyBytes == null) {
int nextValue = nextValue(topic);
List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
if (availablePartitions.size() > 0) {
int part = Utils.toPositive(nextValue) % availablePartitions.size();
return availablePartitions.get(part).partition();
} else {
// no partitions are available, give a non-available partition
return Utils.toPositive(nextValue) % numPartitions;
}
} else {
// hash the keyBytes to choose a partition
return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
}
}
3.1.3 副本（Replication）
同一個 partition 可能會有多個 replication（對應 server.properties 配置中的 default.replication.factor=N）。沒有 replication 的狀況下，一旦b roker 宕機，其上全部 patition 的數據都不可被消費，同時 producer 也不能再將數據存於其上的 partition。引入 replication 以後，同一個 partition 可能會有多個 replication，而這時須要在這些 replication 之間選出一個 leader，producer 和 consumer 只與這個 leader 交互，其它 replication 做爲 follower 從leader 中複製數據。
3.1.4 寫入流程
producer寫入消息流程以下：
1）producer 先從 zookeeper 的 "/brokers/…/state"節點找到該 partition 的 leader
2）producer 將消息發送給該 leader
3）leader 將消息寫入本地 log
4）followers 從 leader pull 消息，寫入本地 log 後向 leader 發送 ACK
5）leader 收到全部ISR中的 replication 的 ACK 後，增長 HW（high watermark，最後 commit 的offset）並向 producer 發送 ACK
注意：要特別注意ACK應答模式！
3.2 Broker 保存消息
3.2.1 存儲方式
物理上把 topic 分紅一個或多個 patition（對應 server.properties 中的num.partitions=3配置），每一個 patition 物理上對應一個文件夾（該文件夾存儲該 patition 的全部消息和索引文件），以下：
[atguigu@hadoop102 logs]$ cd first-0/
[atguigu@hadoop102 first-0]$ ll
總用量 16
-rw-rw-r--. 1 atguigu atguigu 0 3月 4 19:34 00000000000000000000.index
-rw-rw-r--. 1 atguigu atguigu 225 3月 4 18:27 00000000000000000000.log
-rw-rw-r--. 1 atguigu atguigu 12 3月 4 19:34 00000000000000000000.timeindex
-rw-rw-r--. 1 atguigu atguigu 10 3月 4 19:34 00000000000000000003.snapshot
-rw-rw-r--. 1 atguigu atguigu 8 3月 4 18:24 leader-epoch-checkpoint
[atguigu@hadoop102 first-0]$
3.2.2 存儲策略
不管消息是否被消費，kafka 都會保留全部消息。有兩種策略能夠刪除舊數據：
1）基於時間：log.retention.hours=168 （單位是小時，168小時即7天）
2）基於大小：log.retention.bytes=1073741824
須要注意的是，由於 Kafka 讀取特定消息的時間複雜度爲O(1)，即與文件大小無關，因此這裏刪除過時文件與提升 Kafka 性能無關。
3.2.3 Zookeeper存儲結構
注意：producer 不在zk中註冊，消費者在zk中註冊。
3.3 Kafka 消費過程分析
kafka提供了兩套 consumer API：高級 Consumer API 和低級 Consumer API。
3.3.1 高級API
1）高級API優勢
高級 API 寫起來簡單。
不須要自行去管理 offset，系統經過 zookeeper 自行管理。
不須要管理分區、副本等狀況，系統自動管理。
消費者斷線會自動根據上一次記錄在 zookeeper 中的 offset 去接着獲取數據（默認設置1分鐘更新一下 zookeeper 中存的 offset）。
可使用 group 來區分對同一個 topic 的不一樣程序的訪問分離開來（不一樣的 group 記錄不一樣的 offset，這樣不一樣程序讀取同一個 topic 纔不會由於 offset 互相影響）。
2）高級API缺點
不能自行控制offset（對於某些特殊需求來講）。
不能細化控制如分區、副本、zk等。
3.3.2 低級API
1）低級 API 優勢
可以讓開發者本身控制 offset，想從哪裏讀取就從哪裏讀取。
自行控制鏈接分區，對分區自定義進行負載均衡。
對 zookeeper 的依賴性下降（如：offset 不必定非要靠zk存儲，自行存儲 offset 便可，好比存在文件或者內存中）。
2）低級API缺點
太過複雜，須要自行控制 offset，鏈接哪一個分區，找到分區 leader 等。
3.3.3 消費者組
消費者是以 consumer group 消費者組的方式工做，由一個或者多個消費者組成一個組，共同消費一個 topic。每一個分區在同一時間只能由 group 中的一個消費者讀取，可是多個 group 能夠同時消費這個 partition。在圖中，有一個由三個消費者組成的 group，有一個消費者讀取主題中的兩個分區，另外兩個分別讀取一個分區。某個消費者讀取某個分區，也能夠叫作某個消費者是某個分區的擁有者。
在這種狀況下，消費者能夠經過
水平擴展的方式同時讀取大量的消息。另外，若是一個消費者失敗了，那麼其餘的 group 成員會自動負載均衡讀取以前失敗的消費者讀取的分區。
3.3.4 消費方式
consumer採用 pull（拉）模式從 broker 中讀取數據。
push（推）模式很難適應消費速率不一樣的消費者，由於消息發送速率是由 broker 決定的。它的目標是儘量以最快速度傳遞消息，可是這樣很容易形成 consumer 來不及處理消息，典型的表現就是拒絕服務以及網絡擁塞。而 pull 模式則能夠根據 consumer 的消費能力以適當的速率消費消息。
對於 Kafka 而言，pull 模式更合適，它可簡化 broker 的設計，consumer 可自主控制消費消息的速率，同時 consumer 能夠本身控制消費方式--便可批量消費也可逐條消費，同時還能選擇不一樣的提交方式從而實現不一樣的傳輸語義。
pull 模式不足之處是，若是kafka沒有數據，消費者可能會陷入循環中，一直等待數據到達。爲了不這種狀況，咱們在咱們的拉請求中有參數，容許消費者請求在等待數據到達的「長輪詢」中進行阻塞（而且可選地等待到給定的字節數，以確保大的傳輸大小）。
3.3.5 消費者組案例
1）需求：測試同一個消費者組中的消費者，同一時刻只能有一個消費者消費。
2）案例實操：
（1）在hadoop10二、hadoop103上修改/opt/module/kafka/config/consumer.properties配置文件中的group.id屬性爲任意組名。
[atguigu@hadoop103 config]$ vim consumer.properties
group.id=atguigu
（2）在hadoop10二、hadoop103上分別啓動消費者
[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh \
--zookeeper hadoop102:2181 --topic first --consumer.config config/consumer.properties
[atguigu@hadoop103 kafka]$ bin/kafka-console-consumer.sh \
--zookeeper hadoop102:2181 --topic first --consumer.config config/consumer.properties
（3）在hadoop104上啓動生產者
[atguigu@hadoop104 kafka]$ bin/kafka-console-producer.sh \
--broker-list hadoop102:9092 --topic first
>hello world
（4）查看hadoop102和hadoop103的接收者。
結論：同一時刻只有一個消費者接收到消息。
第4章 Kafka API實戰
4.1 環境準備
1）啓動zk集羣和kafka集羣，在kafka集羣中打開一個消費者
[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh \
--zookeeper hadoop102:2181 --topic first
2）導入pom依賴
<dependencies>
<!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>0.11.0.0</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka -->
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.12</artifactId>
<version>0.11.0.2</version>
</dependency>
</dependencies>
4.2 Kafka生產者Java API
4.2.1 建立生產者（過期的API）
package com.atguigu.kafka.producer;
import kafka.producer.KeyedMessage;
import java.util.Properties;
import kafka.javaapi.producer.Producer;
import kafka.producer.ProducerConfig;
/**
* @author chenmingjun
* @date 2019-03-05 11:39
*/
public class OldProducer {
@SuppressWarnings("deprecation")
public static void main(String[] args) {
Properties props = new Properties();
props.put("metadata.broker.list", "hadoop102:9092");
props.put("request.required.acks", "1");
props.put("serializer.class", "kafka.serializer.StringEncoder");
Producer<Integer, String> producer = new Producer<Integer, String>(new ProducerConfig(props));
KeyedMessage<Integer, String> message = new KeyedMessage<Integer, String>("first", "hello world");
producer.send(message);
}
}
4.2.2 建立生產者（新的API）
package com.atguigu.kafka.producer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
/**
* @author chenmingjun
* @date 2019-03-05 11:42
*/
public class NewProducer {
public static void main(String[] args) {
Properties props = new Properties();
// Kafka服務端的主機名和端口號
props.put("bootstrap.servers", "hadoop102:9092");
// 等待全部副本節點的應答（應答級別）all等價於-1
props.put("acks", "all");
// props.put(ProducerConfig.ACKS_CONFIG, "all"); // 兩者等價
// 消息發送最大嘗試次數
props.put("retries", 0);
// 一批消息處理大小
props.put("batch.size", 16384);
// 請求延時
props.put("linger.ms", 1);
// 發送緩存區內存大小（32M）
props.put("buffer.memory", 33554432);
// key序列化
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// value序列化
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// 建立生產者對象
Producer<String, String> producer = new KafkaProducer<String, String>(props);
// 測試循環發送數據
for (int i = 0; i < 50; i++) {
producer.send(new ProducerRecord<String, String>("first", Integer.toString(i), "hello world-" + i));
}
// 關閉資源
producer.close();
}
}
4.2.3 建立生產者帶回調函數（新的API）
package com.atguigu.kafka.producer;
import org.apache.kafka.clients.producer.*;
import java.util.Properties;
import java.util.concurrent.Future;
/**
* @author chenmingjun
* @date 2019-03-05 14:19
*/
public class CallBackNewProducer {
public static void main(String[] args) {
Properties props;
props = new Properties();
// Kafka服務端的主機名和端口號
props.put("bootstrap.servers", "hadoop102:9092");
// 等待全部副本節點的應答（應答級別）all等價於-1
props.put("acks", "all");
// props.put(ProducerConfig.ACKS_CONFIG, "all"); // 兩者等價
// 消息發送最大嘗試次數
props.put("retries", 0);
// 一批消息處理大小
props.put("batch.size", 16384);
// 請求延時
props.put("linger.ms", 1);
// 發送緩存區內存大小（32M）
props.put("buffer.memory", 33554432);
// key序列化
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// value序列化
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// 建立生產者對象
Producer<String, String> producer = new KafkaProducer<String, String>(props);
// 測試循環發送數據
for (int i = 0; i < 50; i++) {
producer.send(new ProducerRecord<String, String>("second", "hello world-" + i), new Callback() {
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
if (metadata != null) {
System.err.println(metadata.partition() + "---" + metadata.offset());
}
}
});
}
// 關閉資源
producer.close();
}
}
4.2.4 自定義分區生產者
0）需求：將全部數據存儲到topic的第0號分區上。
1）定義一個類實現Partitioner接口，重寫裏面的方法（過期API）
package com.atguigu.kafka.producer;
import kafka.producer.Partitioner;
/**
* @author chenmingjun
* @date 2019-03-05 14:47
*/
public class PartitionerOldProducer implements Partitioner {
public PartitionerOldProducer() {
super();
}
@Override
public int partition(Object key, int numPartitions) {
// 控制分區
return 0;
}
}
2）自定義分區（新API）
package com.atguigu.kafka.producer;
import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import java.util.Map;
/**
* @author chenmingjun
* @date 2019-03-05 14:51
*/
public class PartitionerNewProducer implements Partitioner {
@Override
public void configure(Map<String, ?> configs) {
}
@Override
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
// 控制分區
return 0;
}
@Override
public void close() {
}
}
3）在代碼中調用
package com.atguigu.kafka.producer;
import org.apache.kafka.clients.producer.*;
import java.util.Properties;
import java.util.concurrent.Future;
/**
* @author chenmingjun
* @date 2019-03-05 14:19
*/
public class CallBackNewProducer {
public static void main(String[] args) {
Properties props;
props = new Properties();
// Kafka服務端的主機名和端口號
props.put("bootstrap.servers", "hadoop102:9092");
// 等待全部副本節點的應答（應答級別）all等價於-1
props.put("acks", "all");
// props.put(ProducerConfig.ACKS_CONFIG, "all"); // 兩者等價
// 消息發送最大嘗試次數
props.put("retries", 0);
// 一批消息處理大小
props.put("batch.size", 16384);
// 請求延時
props.put("linger.ms", 1);
// 發送緩存區內存大小（32M）
props.put("buffer.memory", 33554432);
// key序列化
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// value序列化
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// 關聯自定義分區
props.put("partitioner.class", "com.atguigu.kafka.producer.PartitionerNewProducer");
// 建立生產者對象
Producer<String, String> producer = new KafkaProducer<String, String>(props);
// 測試循環發送數據
for (int i = 0; i < 50; i++) {
producer.send(new ProducerRecord<String, String>("second", "hello world-" + i), new Callback() {
public void onCompletion(RecordMetadata metadata, Exception exception) {
if (metadata != null) {
System.err.println(metadata.partition() + "---" + metadata.offset());
}
}
});
}
// 關閉資源
producer.close();
}
}
4）測試
（1）在hadoop102上監控/opt/module/kafka/logs/目錄下second主題2個分區的log日誌動態變化狀況
[atguigu@hadoop102 second-0]$ tail -f 00000000000000000000.log
[atguigu@hadoop102 second-1]$ tail -f 00000000000000000000.log
（2）發現數據都存儲到指定的分區了。
4.3 Kafka消費者Java API
4.3.1 高級API
0）在控制檯建立發送者
[atguigu@hadoop104 kafka]$ bin/kafka-console-producer.sh \
--broker-list hadoop102:9092 --topic second
>hello world
1）建立消費者（過期API）
package com.atguigu.kafka.consumer;
import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
/**
* @author chenmingjun
* @date 2019-03-05 15:04
*/
public class OldConsumer {
@SuppressWarnings("deprecation")
public static void main(String[] args) {
Properties props = new Properties();
props.put("zookeeper.connect", "hadoop102:2181");
props.put("group.id", "g1");
props.put("zookeeper.session.timeout.ms", "500");
props.put("zookeeper.sync.time.ms", "250");
props.put("auto.commit.interval.ms", "1000");
// 建立消費者鏈接器
ConsumerConnector consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(props));
HashMap<String, Integer> topicCount = new HashMap<String, Integer>();
topicCount.put("first", 1);
Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCount);
KafkaStream<byte[], byte[]> stream = consumerMap.get("first").get(0);
ConsumerIterator<byte[], byte[]> it = stream.iterator();
while (it.hasNext()) {
System.out.println(new String(it.next().message()));
}
}
}
2）官方提供案例（自動維護消費狀況）（新API）
package com.atguigu.kafka.consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import java.util.Arrays;
import java.util.Collections;
import java.util.Properties;
/**
* @author chenmingjun
* @date 2019-03-05 15:10
*/
public class NewConsumer {
public static void main(String[] args) {
Properties props = new Properties();
// 定義kakfa 服務的地址，不須要將全部broker指定上
props.put("bootstrap.servers", "hadoop102:9092");
// 指定consumer group
props.put("group.id", "test");
// 若是想重複消費topic數據，有三種方式：一、新建一個組。二、使用低級API指定offset。三、使用高級API在不換組的狀況下重複消費topic數據。
// 一、當咱們新建一個組的時候，須要加一個屬性，以下：
// props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
// 是否自動確認offset
props.put("enable.auto.commit", "true");
// 自動確認offset的時間間隔
props.put("auto.commit.interval.ms", "1000");
// key的序列化類
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
// value的序列化類
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
// 定義consumer
KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
// 指定消費者訂閱的topic，可同時訂閱多個
consumer.subscribe(Arrays.asList("first", "second", "third"));
// 三、使用高級API在不換組的狀況下重複消費topic數據。
// consumer.assign(Collections.singletonList(new TopicPartition("second", 0)));
// consumer.seek(new TopicPartition("second", 0), 2);
while (true) {
// 讀取數據，讀取超時時間爲100ms
ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records)
System.out.println(record.topic() + "---" + record.partition() + "---" + record.offset() + "---" + record.value());
}
}
}
4.3.2 低級API
實現使用低級API讀取指定topic，指定partition，指定offset的數據。
1）消費者使用低級API 的主要步驟：
2）方法描述：
3）完整版代碼：
package com.atguigu.kafka.consumer;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import kafka.api.FetchRequest;
import kafka.api.FetchRequestBuilder;
import kafka.api.PartitionOffsetRequestInfo;
import kafka.cluster.BrokerEndPoint;
import kafka.common.ErrorMapping;
import kafka.common.TopicAndPartition;
import kafka.javaapi.FetchResponse;
import kafka.javaapi.OffsetResponse;
import kafka.javaapi.PartitionMetadata;
import kafka.javaapi.TopicMetadata;
import kafka.javaapi.TopicMetadataRequest;
import kafka.javaapi.consumer.SimpleConsumer;
import kafka.message.MessageAndOffset;
/**
* 根據指定的topic、partition、offset來獲取信息
*/
public class SimpleExample {
private List<String> m_replicaBrokers = new ArrayList<String>();
public SimpleExample() {
m_replicaBrokers = new ArrayList<String>();
}
public static void main(String args[]) {
SimpleExample example = new SimpleExample();
// 最大讀取消息數量
long maxReads = Long.parseLong("3");
// 要訂閱的topic
String topic = "second";
// 要查找的分區
int partition = Integer.parseInt("0");
// broker節點的ip，即鏈接kafka集羣
List<String> seeds = new ArrayList<String>();
seeds.add("hadoop102");
seeds.add("hadoop103");
seeds.add("hadoop103");
// 端口
int port = Integer.parseInt("9092");
try {
example.run(maxReads, topic, partition, seeds, port);
} catch (Exception e) {
System.out.println("Oops:" + e);
e.printStackTrace();
}
}
public void run(long a_maxReads, String a_topic, int a_partition, List<String> a_seedBrokers, int a_port) throws Exception {
// 獲取指定Topic partition的元數據
PartitionMetadata metadata = findLeader(a_seedBrokers, a_port, a_topic, a_partition);
if (metadata == null) {
System.out.println("Can't find metadata for Topic and Partition. Exiting");
return;
}
if (metadata.leader() == null) {
System.out.println("Can't find Leader for Topic and Partition. Exiting");
return;
}
String leadBroker = metadata.leader().host();
String clientName = "Client_" + a_topic + "_" + a_partition;
SimpleConsumer consumer = new SimpleConsumer(leadBroker, a_port, 100000, 64 * 1024, clientName);
long readOffset = getLastOffset(consumer, a_topic, a_partition, kafka.api.OffsetRequest.EarliestTime(), clientName);
int numErrors = 0;
while (a_maxReads > 0) {
if (consumer == null) {
consumer = new SimpleConsumer(leadBroker, a_port, 100000, 64 * 1024, clientName);
}
// 建立獲取數據的對象
FetchRequest req = new FetchRequestBuilder().clientId(clientName).addFetch(a_topic, a_partition, readOffset, 100000).build();
// 獲取數據返回值
FetchResponse fetchResponse = consumer.fetch(req);
// 解析返回值
if (fetchResponse.hasError()) {
numErrors++;
// Something went wrong!
short code = fetchResponse.errorCode(a_topic, a_partition);
System.out.println("Error fetching data from the Broker:" + leadBroker + " Reason: " + code);
if (numErrors > 5)
break;
if (code == ErrorMapping.OffsetOutOfRangeCode()) {
// We asked for an invalid offset. For simple case ask for
// the last element to reset
readOffset = getLastOffset(consumer, a_topic, a_partition, kafka.api.OffsetRequest.LatestTime(), clientName);
continue;
}
consumer.close();
consumer = null;
leadBroker = findNewLeader(leadBroker, a_topic, a_partition, a_port);
continue;
}
numErrors = 0;
long numRead = 0;
for (MessageAndOffset messageAndOffset : fetchResponse.messageSet(a_topic, a_partition)) {
long currentOffset = messageAndOffset.offset();
if (currentOffset < readOffset) {
System.out.println("Found an old offset: " + currentOffset + " Expecting: " + readOffset);
continue;
}
readOffset = messageAndOffset.nextOffset();
ByteBuffer payload = messageAndOffset.message().payload();
byte[] bytes = new byte[payload.limit()];
payload.get(bytes);
System.out.println(String.valueOf(messageAndOffset.offset()) + ": " + new String(bytes, "UTF-8"));
numRead++;
a_maxReads--;
}
if (numRead == 0) {
try {
Thread.sleep(1000);
} catch (InterruptedException ie) {
}
}
}
if (consumer != null)
consumer.close();
}
public static long getLastOffset(SimpleConsumer consumer, String topic, int partition, long whichTime, String clientName) {
TopicAndPartition topicAndPartition = new TopicAndPartition(topic, partition);
Map<TopicAndPartition, PartitionOffsetRequestInfo> requestInfo = new HashMap<TopicAndPartition, PartitionOffsetRequestInfo>();
requestInfo.put(topicAndPartition, new PartitionOffsetRequestInfo(whichTime, 1));
kafka.javaapi.OffsetRequest request = new kafka.javaapi.OffsetRequest(requestInfo, kafka.api.OffsetRequest.CurrentVersion(), clientName);
OffsetResponse response = consumer.getOffsetsBefore(request);
if (response.hasError()) {
System.out.println("Error fetching data Offset Data the Broker. Reason: " + response.errorCode(topic, partition));
return 0;
}
long[] offsets = response.offsets(topic, partition);
return offsets[0];
}
private String findNewLeader(String a_oldLeader, String a_topic, int a_partition, int a_port) throws Exception {
for (int i = 0; i < 3; i++) {
boolean goToSleep = false;
PartitionMetadata metadata = findLeader(m_replicaBrokers, a_port, a_topic, a_partition);
if (metadata == null) {
goToSleep = true;
} else if (metadata.leader() == null) {
goToSleep = true;
} else if (a_oldLeader.equalsIgnoreCase(metadata.leader().host()) && i == 0) {
// first time through if the leader hasn't changed give
// ZooKeeper a second to recover
// second time, assume the broker did recover before failover,
// or it was a non-Broker issue
//
goToSleep = true;
} else {
return metadata.leader().host();
}
if (goToSleep) {
Thread.sleep(1000);
}
}
System.out.println("Unable to find new leader after Broker failure. Exiting");
throw new Exception("Unable to find new leader after Broker failure. Exiting");
}
/**
* 尋找leader
* @param a_seedBrokers
* @param a_port
* @param a_topic
* @param a_partition
* @return
*/
private PartitionMetadata findLeader(List<String> a_seedBrokers, int a_port, String a_topic, int a_partition) {
PartitionMetadata returnMetaData = null;
loop:
for (String seed : a_seedBrokers) {
SimpleConsumer consumer = null;
try {
// 建立獲取分區leader的消費者對象
consumer = new SimpleConsumer(seed, a_port, 100000, 64 * 1024, "leaderLookup");
// 建立獲取多個主題元數據信息的請求
List<String> topics = Collections.singletonList(a_topic);
TopicMetadataRequest req = new TopicMetadataRequest(topics);
kafka.javaapi.TopicMetadataResponse resp = consumer.send(req);
// 獲取多個主題的元數據信息
List<TopicMetadata> topicMetadata = resp.topicsMetadata();
for (TopicMetadata topic : topicMetadata) {
// 獲取多個分區的元數據信息
for (PartitionMetadata part : topic.partitionsMetadata()) {
if (part.partitionId() == a_partition) {
returnMetaData = part;
break loop;
}
}
}
} catch (Exception e) {
System.out.println("Error communicating with Broker [" + seed + "] to find Leader for [" + a_topic + ", " + a_partition + "] Reason: " + e);
} finally {
if (consumer != null)
consumer.close();
}
}
if (returnMetaData != null) {
m_replicaBrokers.clear();
for (BrokerEndPoint replica : returnMetaData.replicas()) {
m_replicaBrokers.add(replica.host());
}
}
return returnMetaData;
}
}
第5章 Kafka Producer攔截器(interceptor)
5.1 攔截器原理
Producer 攔截器(interceptor)是在 Kafka 0.10 版本被引入的，主要用於實現 clients 端的定製化控制邏輯。
對於 producer 而言，interceptor 使得用戶在消息發送前以及 producer 回調邏輯前有機會對消息作一些定製化需求，好比修改消息等。同時，producer 容許用戶指定多個 interceptor 按序做用於同一條消息從而造成一個攔截鏈(interceptor chain)。Intercetpor 的實現接口是 org.apache.kafka.clients.producer.ProducerInterceptor，其定義的方法包括：
（1）configure(configs)：
獲取配置信息和初始化數據時調用。
（2）onSend(ProducerRecord)：
該方法封裝進 KafkaProducer.send 方法中，即它運行在用戶主線程中。Producer 確保在消息被序列化以及計算分區前調用該方法。用戶能夠在該方法中對消息作任何操做，但最好保證不要修改消息所屬的 topic 和分區，不然會影響目標分區的計算。
（3）onAcknowledgement(RecordMetadata, Exception)：
該方法會在消息被應答或消息發送失敗時調用，而且一般都是在 producer 回調邏輯觸發以前。onAcknowledgement 運行在 producer 的IO線程中，所以不要在該方法中放入很重的邏輯，不然會拖慢 producer 的消息發送效率。
（4）close：
關閉 interceptor，主要用於執行一些資源清理工做。
如前所述，interceptor 可能被運行在多個線程中，所以在具體實現時用戶須要自行確保線程安全。另外假若指定了多個 interceptor，則 producer 將按照指定順序調用它們，並僅僅是捕獲每一個 interceptor 可能拋出的異常記錄到錯誤日誌中而非在向上傳遞。這在使用過程當中要特別留意。
5.2 攔截器案例
1）需求：
實現一個簡單的雙 interceptor 組成的攔截鏈。第一個 interceptor 會在消息發送前將時間戳信息加到消息 value 的最前部；第二個 interceptor 會在消息發送後更新成功發送消息數或失敗發送消息數。
2）案例實操
（1）增長時間戳攔截器
package com.atguigu.kafka.interceptor;
import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import java.util.Map;
/**
* @author chenmingjun
* @date 2019-03-05 18:56
*/
public class TimeInterceptor implements ProducerInterceptor<String, String> {
public void configure(Map<String, ?> configs) {
}
public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
// 建立一個新的record，把時間戳寫入消息體的最前部
return new ProducerRecord(record.topic(), record.partition(), record.timestamp(), record.key(),
System.currentTimeMillis() + "," + record.value().toString());
}
public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
}
public void close() {
}
}
（2）統計發送消息成功和發送失敗消息數，並在producer關閉時打印這兩個計數器
package com.atguigu.kafka.interceptor;
import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import java.util.Map;
/**
* @author chenmingjun
* @date 2019-03-05 18:59
*/
public class CounterInterceptor implements ProducerInterceptor<String, String> {
private int errorCounter = 0;
private int successCounter = 0;
public void configure(Map<String, ?> configs) {
}
public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
return record;
}
public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
// 統計成功和失敗的次數
if (exception == null) {
successCounter++;
} else {
errorCounter++;
}
}
public void close() {
// 保存結果
System.out.println("Successful sent: " + successCounter);
System.out.println("Failed sent: " + errorCounter);
}
}
（3）producer主程序
package com.atguigu.kafka.interceptor;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
/**
* @author chenmingjun
* @date 2019-03-05 19:03
*/
public class InterceptorProducer {
public static void main(String[] args) throws Exception {
// 1 設置配置信息
Properties props = new Properties();
props.put("bootstrap.servers", "hadoop102:9092");
props.put("acks", "all");
props.put("retries", 0);
props.put("batch.size", 16384);
props.put("linger.ms", 1);
props.put("buffer.memory", 33554432);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// 2 構建攔截鏈
List<String> interceptors = new ArrayList<String>();
interceptors.add("com.atguigu.kafka.interceptor.TimeInterceptor");
interceptors.add("com.atguigu.kafka.interceptor.CounterInterceptor");
props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors);
// 建立生產者對象
String topic = "second";
Producer<String, String> producer = new KafkaProducer<String, String>(props);
// 3 發送消息
for (int i = 0; i < 10; i++) {
ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, "message" + i);
producer.send(record);
}
// 4 必定要關閉producer，這樣纔會調用interceptor的close方法
producer.close();
}
}
3）測試
（1）在kafka上啓動消費者，而後運行客戶端java程序。
[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh \
--zookeeper hadoop102:2181 --from-beginning --topic second
1551784150698,message2
1551784150699,message5
1551784150701,message8
1551784150601,message0
1551784150699,message3
1551784150699,message6
1551784150701,message9
1551784150698,message1
1551784150699,message4
1551784150701,message7
（2）觀察java平臺控制檯輸出數據以下：
Successful sent: 10
Failed sent: 0
第6章 Kafka Streams
6.1 概述
6.1.1 Kafka Streams
Kafka Streams。Apache Kafka開源項目的一個組成部分。是一個功能強大，易於使用的庫。用於在Kafka上構建高可分佈式、拓展性，容錯的應用程序。
6.1.2 Kafka Streams 特色
1）功能強大
高擴展性，彈性，容錯
2）輕量級
無需專門的集羣
一個庫，而不是框架
3）徹底集成
100%的與Kafka 0.10.0版本兼容
易於集成到現有的應用程序
4）實時性
毫秒級延遲
並不是微批處理，而spark是微處理框架
窗口容許亂序數據
容許遲到數據
6.1.3 爲何要有 Kafka Stream？
當前已經有很是多的流式處理系統，最知名且應用最多的開源流式處理系統有Spark Streaming 和 Apache Storm。Apache Storm 發展多年，應用普遍，提供記錄級別的處理能力，當前也支持 SQL on Stream。而 Spark Streaming 基於 Apache Spark，能夠很是方便與圖計算，SQL處理等集成，功能強大，對於熟悉其它 Spark 應用開發的用戶而言使用門檻低。另外，目前主流的 Hadoop 發行版，如 Cloudera 和 Hortonworks，都集成 了Apache Storm 和 Apache Spark，使得部署更容易。
既然 Apache Spark 與 Apache Storm 擁用如此多的優點，那爲什麼還須要 Kafka Stream 呢？主要有以下緣由：
第一，Spark 和 Storm 都是流式處理框架，而Kafka Stream提供的是一個基於Kafka的流式處理類庫。框架要求開發者按照特定的方式去開發邏輯部分，供框架調用。開發者很難了解框架的具體運行方式，從而使得調試成本高，而且使用受限。而 Kafka Stream 做爲流式處理類庫，直接提供具體的類給開發者調用，整個應用的運行方式主要由開發者控制，方便使用和調試。
第二，雖然 Cloudera 與 Hortonworks 方便了 Storm 和 Spark 的部署，可是這些框架的部署仍然相對複雜。而 Kafka Stream 做爲類庫，能夠很是方便的嵌入應用程序中，它對應用的打包和部署基本沒有任何要求。
第三，就流式處理系統而言，基本都支持 Kafka 做爲數據源。例如 Storm 具備專門的 kafka-spout，而 Spark 也提供專門的 spark-streaming-kafka 模塊。事實上，Kafka 基本上是主流的流式處理系統的標準數據源。換言之，大部分流式系統中都已部署了 Kafka，此時使用 Kafka Stream 的成本很是低。
第四，使用 Storm 或 Spark Streaming 時，須要爲框架自己的進程預留資源，如 Storm 的 supervisor 和 Spark on YARN 的 node manager。即便對於應用實例而言，框架自己也會佔用部分資源，如 Spark Streaming 須要爲 shuffle 和 storage 預留內存。可是 Kafka做爲類庫不佔用系統資源。
第五，因爲 Kafka 自己提供數據持久化，所以 Kafka Stream 提供滾動部署和滾動升級以及從新計算的能力。
第六，因爲 Kafka Consumer Rebalance 機制，Kafka Stream 能夠在線動態調整並行度。
6.2 Kafka Stream 數據清洗案例
0）需求：
實時處理單詞帶有」>>>」前綴的內容。例如輸入」atguigu>>>ximenqing」，最終處理成「ximenqing」。
1）需求分析：
2）案例實操
（1）建立一個工程，並添加jar包或在pom文件中添加依賴
<!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-streams -->
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-streams</artifactId>
<version>0.11.0.2</version>
</dependency>
（2）建立主類
package com.atguigu.kafka.stream;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorSupplier;
import org.apache.kafka.streams.processor.TopologyBuilder;
import java.util.Properties;
/**
* @author chenmingjun
* @date 2019-03-05 21:15
*/
public class KafkaStream {
public static void main(String[] args) {
// 定義輸入的topic
String from = "first";
// 定義輸出的topic
String to = "second";
// 設置參數
Properties settings = new Properties();
settings.put(StreamsConfig.APPLICATION_ID_CONFIG, "logFilter");
settings.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");
StreamsConfig config = new StreamsConfig(settings);
// 構建拓撲
TopologyBuilder builder = new TopologyBuilder();
builder.addSource("SOURCE", from).addProcessor("PROCESS", new ProcessorSupplier<byte[], byte[]>() {
@Override
public Processor<byte[], byte[]> get() {
// 具體分析處理
return new LogProcessor();
}
}, "SOURCE")
.addSink("SINK", to, "PROCESS");
// 建立kafka stream
KafkaStreams streams = new KafkaStreams(builder, config);
streams.start();
}
}
（3）具體業務處理
package com.atguigu.kafka.stream;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
/**
* @author chenmingjun
* @date 2019-03-05 21:26
*/
public class LogProcessor implements Processor<byte[], byte[]> {
private ProcessorContext context;
@Override
public void init(ProcessorContext context) {
this.context = context;
}
@Override
public void process(byte[] key, byte[] value) {
String input = new String(value);
// 若是包含「>>>」則只保留該標記後面的內容
if (input.contains(">>>")) {
input = input.split(">>>")[1].trim();
// 輸出到下一個topic
context.forward("logProcessor".getBytes(), input.getBytes());
} else {
context.forward("logProcessor".getBytes(), input.getBytes());
}
}
@Override
public void punctuate(long l) {
}
@Override
public void close() {
}
}
（4）運行程序
（5）在hadoop104上啓動生產者
[atguigu@hadoop104 kafka]$ bin/kafka-console-producer.sh \
--broker-list hadoop102:9092 --topic first
>hello>>>world
>h>>>atguigu
>hahaha
（6）在hadoop103上啓動消費者
[atguigu@hadoop103 kafka]$ bin/kafka-console-consumer.sh \
--zookeeper hadoop102:2181 --from-beginning --topic second
world
atguigu
hahaha
第7章 擴展知識
7.1 Kafka 與 Flume 比較
在企業中必需要清楚流式數據採集框架 flume 和 kafka 的定位是什麼：
-
flume：Cloudera 公司研發：
- 適合多個生產者；（一個生產者對應一個 Agent 任務）
- 適合下游數據消費者很少的狀況；（多 channel 多 sink 會耗費不少內存）
- 適合數據安全性要求不高的操做；（實際中更多使用 Memory Channel）
- 適合與 Hadoop 生態圈對接的操做。（Cloudera 公司的特長）
-
kafka：Linkedin 公司研發：
- 適合數據下游消費者衆多的狀況；（開啓更多的消費者任務便可，與 Kafka 集羣無關）
- 適合數據安全性要求較高的操做，支持replication。（數據放在磁盤裏）
-
所以咱們經常使用的一種模型是：
- 線上數據 --> flume(適合採集tomcat日誌) --> kafka(離線/實時) --> flume(根據情景增刪該流程) --> HDFS
7.2 Flume 與 kafka 集成
1）配置flume(flume-kafka.conf)
# define
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# source
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F -c +0 /opt/module/datas/flume.log
a1.sources.r1.shell = /bin/bash -c
# sink
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.bootstrap.servers = hadoop102:9092,hadoop103:9092,hadoop104:9092
a1.sinks.k1.kafka.topic = first
a1.sinks.k1.kafka.flumeBatchSize = 20
a1.sinks.k1.kafka.producer.acks = 1
a1.sinks.k1.kafka.producer.linger.ms = 1
# channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# bind
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
2） 啓動kafka IDEA消費者
3） 進入flume根目錄下，啓動flume
[atguigu@hadoop102 flume]$ bin/flume-ng agent -n a1 -c conf/ -f job/flume-kafka.conf
4） 向 /opt/module/datas/flume.log裏追加數據，查看kafka消費者消費狀況
[atguigu@hadoop102 datas]$$ echo hello > /opt/module/datas/flume.log
7.3 Kafka配置信息
7.3.1 Broker 配置信息
| 屬性 | 默認值 | 描述 |
|---|---|---|
broker.id |
必填參數，broker的惟一標識 |
|
log.dirs |
/tmp/kafka-logs |
Kafka 數據存放的目錄。能夠指定多個目錄，中間用逗號分隔，當新partition被建立的時會被存放到當前存放partition最少的目錄。 |
port |
9092 |
BrokerServer接受客戶端鏈接的端口號 |
zookeeper.connect |
null |
Zookeeper的鏈接串，格式爲：hostname1:port1,hostname2:port2,hostname3:port3。能夠填一個或多個，爲了提升可靠性，建議都填上。注意，此配置容許咱們指定一個zookeeper路徑來存放此kafka集羣的全部數據，爲了與其餘應用集羣區分開，建議在此配置中指定本集羣存放目錄，格式爲：hostname1:port1,hostname2:port2,hostname3:port3/chroot/path 。須要注意的是，消費者的參數要和此參數一致。 |
| message.max.bytes | 1000000 | 服務器能夠接收到的最大的消息大小。注意此參數要和consumer的maximum.message.size大小一致，不然會由於生產者生產的消息太大致使消費者沒法消費。 |
| num.io.threads | 8 | 服務器用來執行讀寫請求的IO線程數，此參數的數量至少要等於服務器上磁盤的數量。 |
| queued.max.requests | 500 | I/O線程能夠處理請求的隊列大小，若實際請求數超過此大小，網絡線程將中止接收新的請求。 |
| socket.send.buffer.bytes | 100 * 1024 | The SO_SNDBUFF buffer the server prefers for socket connections. |
| socket.receive.buffer.bytes | 100 * 1024 | The SO_RCVBUFF buffer the server prefers for socket connections. |
| socket.request.max.bytes | 100 * 1024 * 1024 | 服務器容許請求的最大值， 用來防止內存溢出，其值應該小於 Java heap size. |
| num.partitions | 1 | 默認partition數量，若是topic在建立時沒有指定partition數量，默認使用此值，建議改成5 |
| log.segment.bytes | 1024 * 1024 * 1024 | Segment文件的大小，超過此值將會自動新建一個segment，此值能夠被topic級別的參數覆蓋。 |
| log.roll.{ms,hours} | 24 * 7 hours | 新建segment文件的時間，此值能夠被topic級別的參數覆蓋。 |
| log.retention.{ms,minutes,hours} | 7 days | Kafka segment log的保存週期，保存週期超過此時間日誌就會被刪除。此參數能夠被topic級別參數覆蓋。數據量大時，建議減少此值。 |
| log.retention.bytes | -1 | 每一個partition的最大容量，若數據量超過此值，partition數據將會被刪除。注意這個參數控制的是每一個partition而不是topic。此參數能夠被log級別參數覆蓋。 |
| log.retention.check.interval.ms | 5 minutes | 刪除策略的檢查週期 |
| auto.create.topics.enable | true | 自動建立topic參數，建議此值設置爲false，嚴格控制topic管理，防止生產者錯寫topic。 |
| default.replication.factor | 1 | 默認副本數量，建議改成2。 |
| replica.lag.time.max.ms | 10000 | 在此窗口時間內沒有收到follower的fetch請求，leader會將其從ISR(in-sync replicas)中移除。 |
| replica.lag.max.messages | 4000 | 若是replica節點落後leader節點此值大小的消息數量，leader節點就會將其從ISR中移除。 |
| replica.socket.timeout.ms | 30 * 1000 | replica向leader發送請求的超時時間。 |
| replica.socket.receive.buffer.bytes | 64 * 1024 | The socket receive buffer for network requests to the leader for replicating data. |
| replica.fetch.max.bytes | 1024 * 1024 | The number of byes of messages to attempt to fetch for each partition in the fetch requests the replicas send to the leader. |
| replica.fetch.wait.max.ms | 500 | The maximum amount of time to wait time for data to arrive on the leader in the fetch requests sent by the replicas to the leader. |
| num.replica.fetchers | 1 | Number of threads used to replicate messages from leaders. Increasing this value can increase the degree of I/O parallelism in the follower broker. |
| fetch.purgatory.purge.interval.requests | 1000 | The purge interval (in number of requests) of the fetch request purgatory. |
| zookeeper.session.timeout.ms | 6000 | ZooKeeper session 超時時間。若是在此時間內server沒有向zookeeper發送心跳，zookeeper就會認爲此節點已掛掉。 此值過低致使節點容易被標記死亡；若過高，.會致使太遲發現節點死亡。 |
| zookeeper.connection.timeout.ms | 6000 | 客戶端鏈接zookeeper的超時時間。 |
| zookeeper.sync.time.ms | 2000 | H ZK follower落後 ZK leader的時間。 |
| controlled.shutdown.enable | true | 容許broker shutdown。若是啓用，broker在關閉本身以前會把它上面的全部leaders轉移到其它brokers上，建議啓用，增長集羣穩定性。 |
| auto.leader.rebalance.enable | true | If this is enabled the controller will automatically try to balance leadership for partitions among the brokers by periodically returning leadership to the 「preferred」 replica for each partition if it is available. |
| leader.imbalance.per.broker.percentage | 10 | The percentage of leader imbalance allowed per broker. The controller will rebalance leadership if this ratio goes above the configured value per broker. |
| leader.imbalance.check.interval.seconds | 300 | The frequency with which to check for leader imbalance. |
| offset.metadata.max.bytes | 4096 | The maximum amount of metadata to allow clients to save with their offsets. |
| connections.max.idle.ms | 600000 | Idle connections timeout: the server socket processor threads close the connections that idle more than this. |
| num.recovery.threads.per.data.dir | 1 | The number of threads per data directory to be used for log recovery at startup and flushing at shutdown. |
| unclean.leader.election.enable | true | Indicates whether to enable replicas not in the ISR set to be elected as leader as a last resort, even though doing so may result in data loss. |
delete.topic.enable |
false |
啓用deletetopic參數，建議設置爲true。 |
| offsets.topic.num.partitions | 50 | The number of partitions for the offset commit topic. Since changing this after deployment is currently unsupported, we recommend using a higher setting for production (e.g., 100-200). |
| offsets.topic.retention.minutes | 1440 | Offsets that are older than this age will be marked for deletion. The actual purge will occur when the log cleaner compacts the offsets topic. |
| offsets.retention.check.interval.ms | 600000 | The frequency at which the offset manager checks for stale offsets. |
| offsets.topic.replication.factor | 3 | The replication factor for the offset commit topic. A higher setting (e.g., three or four) is recommended in order to ensure higher availability. If the offsets topic is created when fewer brokers than the replication factor then the offsets topic will be created with fewer replicas. |
| offsets.topic.segment.bytes | 104857600 | Segment size for the offsets topic. Since it uses a compacted topic, this should be kept relatively low in order to facilitate faster log compaction and loads. |
| offsets.load.buffer.size | 5242880 | An offset load occurs when a broker becomes the offset manager for a set of consumer groups (i.e., when it becomes a leader for an offsets topic partition). This setting corresponds to the batch size (in bytes) to use when reading from the offsets segments when loading offsets into the offset manager’s cache. |
| offsets.commit.required.acks | -1 | The number of acknowledgements that are required before the offset commit can be accepted. This is similar to the producer’s acknowledgement setting. In general, the default should not be overridden. |
| offsets.commit.timeout.ms | 5000 | The offset commit will be delayed until this timeout or the required number of replicas have received the offset commit. This is similar to the producer request timeout. |
7.3.2 Producer 配置信息
| 屬性 | 默認值 | 描述 |
|---|---|---|
| metadata.broker.list | 啓動時producer查詢brokers的列表，能夠是集羣中全部brokers的一個子集。注意，這個參數只是用來獲取topic的元信息用，producer會從元信息中挑選合適的broker並與之創建socket鏈接。格式是：host1:port1,host2:port2。 | |
| request.required.acks | 0 | 參見3.2節介紹 |
| request.timeout.ms | 10000 | Broker等待ack的超時時間，若等待時間超過此值，會返回客戶端錯誤信息。 |
| producer.type | sync | 同步異步模式。async表示異步，sync表示同步。若是設置成異步模式，能夠容許生產者以batch的形式push數據，這樣會極大的提升broker性能，推薦設置爲異步。 |
| serializer.class | kafka.serializer.DefaultEncoder | 序列號類，.默認序列化成 byte[] 。 |
| key.serializer.class | Key的序列化類，默認同上。 | |
| partitioner.class | kafka.producer.DefaultPartitioner | Partition類，默認對key進行hash。 |
| compression.codec | none | 指定producer消息的壓縮格式，可選參數爲： 「none」, 「gzip」 and 「snappy」。關於壓縮參見4.1節 |
| compressed.topics | null | 啓用壓縮的topic名稱。若上面參數選擇了一個壓縮格式，那麼壓縮僅對本參數指定的topic有效，若本參數爲空，則對全部topic有效。 |
| message.send.max.retries | 3 | Producer發送失敗時重試次數。若網絡出現問題，可能會致使不斷重試。 |
| retry.backoff.ms | 100 | Before each retry, the producer refreshes the metadata of relevant topics to see if a new leader has been elected. Since leader election takes a bit of time, this property specifies the amount of time that the producer waits before refreshing the metadata. |
| topic.metadata.refresh.interval.ms | 600 * 1000 | The producer generally refreshes the topic metadata from brokers when there is a failure (partition missing, leader not available…). It will also poll regularly (default: every 10min so 600000ms). If you set this to a negative value, metadata will only get refreshed on failure. If you set this to zero, the metadata will get refreshed after each message sent (not recommended). Important note: the refresh happen only AFTER the message is sent, so if the producer never sends a message the metadata is never refreshed |
| queue.buffering.max.ms | 5000 | 啓用異步模式時，producer緩存消息的時間。好比咱們設置成1000時，它會緩存1秒的數據再一次發送出去，這樣能夠極大的增長broker吞吐量，但也會形成時效性的下降。 |
| queue.buffering.max.messages | 10000 | 採用異步模式時producer buffer 隊列裏最大緩存的消息數量，若是超過這個數值，producer就會阻塞或者丟掉消息。 |
| queue.enqueue.timeout.ms | -1 | 當達到上面參數值時producer阻塞等待的時間。若是值設置爲0，buffer隊列滿時producer不會阻塞，消息直接被丟掉。若值設置爲-1，producer會被阻塞，不會丟消息。 |
| batch.num.messages | 200 | 採用異步模式時，一個batch緩存的消息數量。達到這個數量值時producer纔會發送消息。 |
| send.buffer.bytes | 100 * 1024 | Socket write buffer size |
| client.id | 「」 | The client id is a user-specified string sent in each request to help trace calls. It should logically identify the application making the request. |
7.3.3 Consumer 配置信息
| 屬性 | 默認值 | 描述 |
|---|---|---|
| group.id | Consumer的組ID，相同goup.id的consumer屬於同一個組。 | |
| zookeeper.connect | Consumer的zookeeper鏈接串，要和broker的配置一致。 | |
| consumer.id | null | 若是不設置會自動生成。 |
| socket.timeout.ms | 30 * 1000 | 網絡請求的socket超時時間。實際超時時間由max.fetch.wait + socket.timeout.ms 肯定。 |
| socket.receive.buffer.bytes | 64 * 1024 | The socket receive buffer for network requests. |
| fetch.message.max.bytes | 1024 * 1024 | 查詢topic-partition時容許的最大消息大小。consumer會爲每一個partition緩存此大小的消息到內存，所以，這個參數能夠控制consumer的內存使用量。這個值應該至少比server容許的最大消息大小大，以避免producer發送的消息大於consumer容許的消息。 |
| num.consumer.fetchers | 1 | The number fetcher threads used to fetch data. |
| auto.commit.enable | true | 若是此值設置爲true，consumer會週期性的把當前消費的offset值保存到zookeeper。當consumer失敗重啓以後將會使用此值做爲新開始消費的值。 |
| auto.commit.interval.ms | 60 * 1000 | Consumer提交offset值到zookeeper的週期。 |
| queued.max.message.chunks | 2 | 用來被consumer消費的message chunks 數量， 每一個chunk能夠緩存fetch.message.max.bytes大小的數據量。 |
| auto.commit.interval.ms | 60 * 1000 | Consumer提交offset值到zookeeper的週期。 |
| queued.max.message.chunks | 2 | 用來被consumer消費的message chunks 數量， 每一個chunk能夠緩存fetch.message.max.bytes大小的數據量。 |
| fetch.min.bytes | 1 | The minimum amount of data the server should return for a fetch request. If insufficient data is available the request will wait for that much data to accumulate before answering the request. |
| fetch.wait.max.ms | 100 | The maximum amount of time the server will block before answering the fetch request if there isn’t sufficient data to immediately satisfy fetch.min.bytes. |
| rebalance.backoff.ms | 2000 | Backoff time between retries during rebalance. |
| refresh.leader.backoff.ms | 200 | Backoff time to wait before trying to determine the leader of a partition that has just lost its leader. |
| auto.offset.reset | largest | What to do when there is no initial offset in ZooKeeper or if an offset is out of range ;smallest : automatically reset the offset to the smallest offset; largest : automatically reset the offset to the largest offset;anything else: throw exception to the consumer |
| consumer.timeout.ms | -1 | 若在指定時間內沒有消息消費，consumer將會拋出異常。 |
| exclude.internal.topics | true | Whether messages from internal topics (such as offsets) should be exposed to the consumer. |
| zookeeper.session.timeout.ms | 6000 | ZooKeeper session timeout. If the consumer fails to heartbeat to ZooKeeper for this period of time it is considered dead and a rebalance will occur. |
| zookeeper.connection.timeout.ms | 6000 | The max time that the client waits while establishing a connection to zookeeper. |
| zookeeper.sync.time.ms | 2000 | How far a ZK follower can be behind a ZK leader |
7.4 如何查看 Kafka 集羣維護的 offset 信息
步驟：
（1）修改配置文件consumer.properties，增長一個屬性
[atguigu@hadoop102 config]$ pwd
/opt/module/kafka/config
[atguigu@hadoop102 config]$ vim consumer.properties
exclude.internal.topics=false
（2）分發配置好的額文件
[atguigu@hadoop102 config]$ xsync consumer.properties
（3）執行新的消費者命令
bin/kafka-console-consumer.sh \
--zookeeper hadoop102:2181 --topic __consumer_offsets \
--formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" \
--consumer.config config/consumer.properties \
--from-beginning
相關文章
- 1. 大數據技術之_10_Kafka學習_Kafka概述+Kafka集羣部署+Kafka工做流程分析+Kafka API實戰+Kafka Producer攔截器+Kafka Streams
- 2. Kafka Producer攔截器
- 3. 大數據技術之Kafka第2章 Kafka集羣部署
- 4. Kafka producer攔截器與 Stream
- 5. Kafka中Producer攔截器
- 6. Kafka producer攔截器(interceptor)
- 7. 大數據技術之Kafka 第6章 kafka Streams
- 8. Kafka攔截器
- 9. Kafka集羣部署
- 10. Kafka（五）Kafka的API操做和攔截器
- 更多相關文章...
- • Git 工作流程 - Git 教程
- • Swarm 集羣管理 - Docker教程
- • Java Agent入門實戰（二）-Instrumentation源碼概述
- • Docker容器實戰(一) - 封神Server端技術
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
歡迎關注本站公眾號,獲取更多信息
相關文章
- 1. 大數據技術之_10_Kafka學習_Kafka概述+Kafka集羣部署+Kafka工做流程分析+Kafka API實戰+Kafka Producer攔截器+Kafka Streams
- 2. Kafka Producer攔截器
- 3. 大數據技術之Kafka第2章 Kafka集羣部署
- 4. Kafka producer攔截器與 Stream
- 5. Kafka中Producer攔截器
- 6. Kafka producer攔截器(interceptor)
- 7. 大數據技術之Kafka 第6章 kafka Streams
- 8. Kafka攔截器
- 9. Kafka集羣部署
- 10. Kafka（五）Kafka的API操做和攔截器