第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 資訊
第1章 Kafka概述
1.1 消息隊列
1）點對點模式（一對一，消費者主動
拉取
資料，消息收到後消息清除）
點對點模型通常是一個
基于拉取或者輪詢
的消息傳送模型，這種模型從隊列中請求資訊，而不是将消息推送到用戶端。這個模型的特點是發送到隊列的消息被一個且隻有一個接收者接收處理，即使有多個消息監聽者也是如此。
2）釋出/訂閱模式（一對多，資料生産後，推送給所有訂閱者）
釋出訂閱模型則是一個
基于推送
的消息傳送模型。釋出訂閱模型可以有多種不同的訂閱者，臨時訂閱者隻在主動監聽主題時才接收消息，而持久訂閱者則監聽主題的所有消息，即使目前訂閱者不可用，處于離線狀态。
1.2 為什麼需要消息隊列
1）解耦：
允許你獨立的擴充或修改兩邊的處理過程，隻要確定它們遵守同樣的接口限制。
2）備援：
消息隊列把資料進行持久化直到它們已經被完全處理，通過這一方式規避了資料丢失風險。許多消息隊列所采用的"插入-擷取-删除"範式中，在把一個消息從隊列中删除之前，需要你的處理系統明确的指出該消息已經被處理完畢，進而確定你的資料被安全的儲存直到你使用完畢。
3）擴充性：
因為消息隊列解耦了你的處理過程，是以增大消息入隊和處理的頻率是很容易的，隻要另外增加處理過程即可。
4）靈活性 & 峰值處理能力：
在通路量劇增的情況下，應用仍然需要繼續發揮作用，但是這樣的突發流量并不常見。如果為以能處理這類峰值通路為标準來投入資源随時待命無疑是巨大的浪費。使用消息隊列能夠使關鍵元件頂住突發的通路壓力，而不會因為突發的超負荷的請求而完全崩潰。
5）可恢複性：
系統的一部分元件失效時，不會影響到整個系統。消息隊列降低了程序間的耦合度，是以即使一個處理消息的程序挂掉，加入隊列中的消息仍然可以在系統恢複後被處理。
6）順序保證：
在大多使用場景下，資料處理的順序都很重要。大部分消息隊列本來就是排序的，并且能保證資料會按照特定的順序來處理。（Kafka保證一個Partition内的消息的有序性）
7）緩沖：
有助于控制和優化資料流經過系統的速度，解決生産消息和消費消息的處理速度不一緻的情況。
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畢業，并成為***開源項目。該項目的目标是為處理實時資料提供一個統一、高通量、低等待的平台。
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=1、broker.id=2
注：
broker.id不得重複。
8）啟動Kafka叢集
依次在hadoop102、hadoop103、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 config/server.properties
[atguigu@hadoop103 kafka]$ bin/kafka-server-stop.sh -daemon config/server.properties
[atguigu@hadoop104 kafka]$ bin/kafka-server-stop.sh -daemon config/server.properties
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）在hadoop102、hadoop103上修改/opt/module/kafka/config/consumer.properties配置檔案中的group.id屬性為任意組名。
[atguigu@hadoop103 config]$ vim consumer.properties
group.id=atguigu
（2）在hadoop102、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資料，有三種方式：1、建立一個組。2、使用低級API指定offset。3、使用進階API在不換組的情況下重複消費topic資料。
// 1、當我們建立一個組的時候，需要加一個屬性，如下：
// 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"));
// 3、使用進階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 配置資訊
| 屬性 | 預設值 | 描述 |
|---|---|---|
| | |
| | |
| | |
| | |
| 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. |
| | |
| 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 | 參見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