kafka生產調優手冊

一、Kafka 硬體 配置選擇

1、場景說明

2、服務器臺數選擇

3、磁盤選擇

4、記憶體選擇

Kafka 記憶體組成：堆記憶體 + 頁快取

1）Kafka 堆記憶體建議每個節點：10g ~ 15g

在 kafka-server-start.sh 中修改

if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
	export KAFKA_HEAP_OPTS="-Xmx10G -Xms10G"
fi

查看 Kafka 行程號:

[atguigu@hadoop102 kafka]$ jps
2321 Kafka
5255 Jps
1931 QuorumPeerMain

根據 Kafka 行程號，查看 Kafka 的 GC 情況:

jstat -gc 2321 1s 10

新生代GC次數


根據 Kafka 行程號，查看 Kafka 的堆記憶體：

[atguigu@hadoop102 kafka]$ jmap -heap 2321

Attaching to process ID 2321, please wait...
Debugger attached successfully.
Server compiler detected.
JVM version is 25.212-b10

using thread-local object allocation.
Garbage-First (G1) GC with 8 thread(s)

Heap Configuration:
	MinHeapFreeRatio = 40
	MaxHeapFreeRatio = 70
	MaxHeapSize = 2147483648 (2048.0MB)
	NewSize = 1363144 (1.2999954223632812MB)
	MaxNewSize = 1287651328 (1228.0MB)
	OldSize = 5452592 (5.1999969482421875MB)
	NewRatio = 2
	SurvivorRatio = 8
	MetaspaceSize = 21807104 (20.796875MB)
	CompressedClassSpaceSize = 1073741824 (1024.0MB)
	MaxMetaspaceSize = 17592186044415 MB
	G1HeapRegionSize = 1048576 (1.0MB)

Heap Usage:
G1 Heap:
	regions = 2048
	capacity = 2147483648 (2048.0MB)
	used = 246367744 (234.95458984375MB)
	free = 1901115904 (1813.04541015625MB)
	11.472392082214355% used

G1 Young Generation:
Eden Space:
	regions = 83
	capacity = 105906176 (101.0MB)
	used = 87031808 (83.0MB)
	free = 18874368 (18.0MB)
	82.17821782178218% used

Survivor Space:
	regions = 7
	capacity = 7340032 (7.0MB)
	used = 7340032 (7.0MB)
	free = 0 (0.0MB)
	100.0% used

G1 Old Generation:
	regions = 147
	capacity = 2034237440 (1940.0MB)
	used = 151995904 (144.95458984375MB)
	free = 1882241536 (1795.04541015625MB)
	7.471886074420103% used

13364 interned Strings occupying 1449608 bytes.

頁快取：頁快取是 Linux 系統服務器的記憶體，

我們只需要保證 1 個 segment（1g）中25%的資料在記憶體中就好，

5、CPU 選擇

6、網路選擇

二、Kafka 生產者

1、Kafka 生產者核心引數配置

2、生產者如何提高吞吐量

3、資料可靠性

4、資料去重

5、資料有序

單磁區內，有序（有條件的，不能亂序）；

多磁區，磁區與磁區間無序；

6、資料 亂序

三、Kafka Broker

1、Broker 核心引數配置

2、服役新節點/ 退役舊節點

（1）創建一個要均衡的主題，

[atguigu@hadoop102 kafka]$ vim topics-to-move.json
{
	"topics": [
		{"topic": "first"}
	],
	"version": 1
}

（2）生成一個負載均衡的計劃，

bin/kafka-reassign-partitions.sh --bootstrap-server  hadoop102:9092  --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2,3" --generate

（3）創建副本存盤計劃（所有副本存盤在 broker0、broker1、broker2、broker3 中），

vim increase-replication-factor.json

（4）執行副本存盤計劃，

bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --execute

（5）驗證副本存盤計劃，

bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --verify

3、增加磁區

修改磁區數（注意：磁區數只能增加，不能減少）

bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --alter --topic first --partitions 3

4、增加副本因子

1）創建 topic

bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --partitions 3 --replication-factor 1 --topic four

2）手動增加副本存盤

5、手動調整磁區副本存盤

（1）創建副本存盤計劃（所有副本都指定存盤在 broker0、broker1 中），

vim increase-replication-factor.json

輸入如下內容：

{
	"version":1,
	"partitions":[{"topic":"three","partition":0,"replicas":[0,1]},
	{"topic":"three","partition":1,"replicas":[0,1]},
	{"topic":"three","partition":2,"replicas":[1,0]},
	{"topic":"three","partition":3,"replicas":[1,0]}]
}

（2）執行副本存盤計劃，

bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --execute

（3）驗證副本存盤計劃，

bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --verify

6、Leader Partition 負載 平衡

7、自動創建主題

1）向一個沒有提前創建 five 主題發送資料

bin/kafka-console-producer.sh  --bootstrap-server hadoop102:9092 --topic five
>hello world

2）查看 five 主題的詳情

bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic five

四、 Kafka 消費者

1、Kafka 消費者核心引數配置

2、消費者再平衡

3、指定 Offset 消費

kafkaConsumer.seek(topic, 1000);

4、指定 時間 消費

HashMap<TopicPartition, Long> timestampToSearch = new HashMap<>();
timestampToSearch.put(topicPartition, System.currentTimeMillis() - 1 * 24 * 3600 * 1000);
kafkaConsumer.offsetsForTimes(timestampToSearch);

5、消費者如何提高吞吐量

增加磁區數；

bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --alter --topic first --partitions 3

五、Kafka 總體

1、如何提升吞吐量

2、資料精準一次

3、合理設定 磁區數

4、單條日志大于 1m

5、服務器掛了

6、集群壓力測驗

1 ）Kafka 壓測

用 Kafka 官方自帶的腳本，對 Kafka 進行壓測，
? 生產者壓測：kafka-producer-perf-test.sh
? 消費者壓測：kafka-consumer-perf-test.sh

2 ）Kafka Producer 壓力測驗

創建一個 test topic，設定為 3 個磁區 3 個副本

bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --replication-factor 3 --partitions 3 --topic test

在/opt/module/kafka/bin 目錄下面有這兩個檔案，我們來測驗一下

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

引數說明：

輸出結果：

（3）調整 batch.size 大小

①batch.size 默認值是 16k，本次實驗 batch.size 設定為32k，

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

輸出結果：

②batch.size 默認值是 16k，本次實驗 batch.size 設定為 4k，

bin/kafka-producer-perf-test.sh -- topic test --record-size 1024 --num-records 1000000 --throughput 10000 

bootstrap.servers=hadoop102:9092,hadoop103:9092,hadoop104:9092 batch.size=4096 linger.ms=0

輸出結果：

（4）調整 linger.ms 時間

（5）調整壓縮方式




（6）調整快取大小

3 ）Kafka Consumer 壓力測驗

引數說明：

（3）一次拉取條數為 2000

（4）調整 fetch.max.bytes 大小為 100m