Topic 級別參數(shù)

你可能會有這樣的疑問：如果同時設(shè)置了 Topic 級別參數(shù)和全局 Broker 參數(shù)，到底聽誰的呢？哪個說了算呢？答案就是 Topic 級別參數(shù)會覆蓋全局 Broker 參數(shù)的值，而每個 Topic 都能設(shè)置自己的參數(shù)值，這就是所謂的 Topic 級別參數(shù)。

比如消息數(shù)據(jù)的留存時間參數(shù)，在實際生產(chǎn)環(huán)境中，如果為所有 Topic 的數(shù)據(jù)都保存相當(dāng)長的時間，這樣做既不高效也無必要。更適當(dāng)?shù)淖龇ㄊ窃试S不同部門的 Topic 根據(jù)自身業(yè)務(wù)需要，設(shè)置自己的留存時間。如果只能設(shè)置全局 Broker 參數(shù)，那么勢必要提取所有業(yè)務(wù)留存時間的最大值作為全局參數(shù)值，此時設(shè)置 Topic 級別參數(shù)把它覆蓋，就是一個不錯的選擇。

從保存消息方面來考量的話，下面這組參數(shù)是非常重要的：

• retention.ms：規(guī)定了該 Topic 消息被保存的時長。默認(rèn)是 7 天，即該 Topic 只保存最近 7 天的消息。一旦設(shè)置了這個值，它會覆蓋掉 Broker 端的全局參數(shù)值。
• retention.bytes：規(guī)定了要為該 Topic 預(yù)留多大的磁盤空間。和全局參數(shù)作用相似，這個值通常在多租戶的 Kafka 集群中會有用武之地。當(dāng)前默認(rèn)值是 -1，表示可以無限使用磁盤空間。

上面這些是從保存消息的維度來說的。如果從能處理的消息大小這個角度來看的話，有一個參數(shù)是必須要設(shè)置的，即max.message.bytes。它決定了 Kafka Broker 能夠正常接收該 Topic 的最大消息大小。我知道目前在很多公司都把 Kafka 作為一個基礎(chǔ)架構(gòu)組件來運(yùn)行，上面跑了很多的業(yè)務(wù)數(shù)據(jù)。如果在全局層面上，我們不好給出一個合適的最大消息值，那么不同業(yè)務(wù)部門能夠自行設(shè)定這個 Topic 級別參數(shù)就顯得非常必要了。在實際場景中，這種用法也確實是非常常見的。

Topic 級別參數(shù)的設(shè)置就是這種情況，我們有兩種方式可以設(shè)置：

• 創(chuàng)建 Topic 時進(jìn)行設(shè)置
• 修改 Topic 時設(shè)置

我們先來看看如何在創(chuàng)建 Topic 時設(shè)置這些參數(shù)。我用上面提到的retention.ms和max.message.bytes舉例。設(shè)想你的部門需要將交易數(shù)據(jù)發(fā)送到 Kafka 進(jìn)行處理，需要保存最近半年的交易數(shù)據(jù)，同時這些數(shù)據(jù)很大，通常都有幾 MB，但一般不會超過 5MB。現(xiàn)在讓我們用以下命令來創(chuàng)建 Topic：

bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic transaction --partitions 1 --replication-factor 1 --config retention.ms=15552000000 --config max.message.bytes=5242880

我們只需要知道 Kafka 開放了kafka-topics命令供我們來創(chuàng)建 Topic 即可。對于上面這樣一條命令，請注意結(jié)尾處的--config設(shè)置，我們就是在 config 后面指定了想要設(shè)置的 Topic 級別參數(shù)。

下面看看使用另一個自帶的命令kafka-configs來修改 Topic 級別參數(shù)。假設(shè)我們現(xiàn)在要發(fā)送最大值是 10MB 的消息，該如何修改呢？命令如下：

bin/kafka-configs.sh --zookeeper localhost:2181 --entity-type topics --entity-name transaction --alter --add-config max.message.bytes=10485760

總體來說，你只能使用這么兩種方式來設(shè)置 Topic 級別參數(shù)。我個人的建議是，你最好始終堅持使用第二種方式來設(shè)置，并且在未來，Kafka 社區(qū)很有可能統(tǒng)一使用kafka-configs腳本來調(diào)整 Topic 級別參數(shù)。

JVM 參數(shù)

Kafka 服務(wù)器端代碼是用 Scala 語言編寫的，但終歸還是編譯成 Class 文件在 JVM 上運(yùn)行，因此 JVM 參數(shù)設(shè)置對于 Kafka 集群的重要性不言而喻

首先我先說說 Java 版本，我個人極其不推薦將 Kafka 運(yùn)行在 Java 6 或 7 的環(huán)境上。Java 6 實在是太過陳舊了，沒有理由不升級到更新版本。另外 Kafka 自 2.0.0 版本開始，已經(jīng)正式摒棄對 Java 7 的支持了，所以有條件的話至少使用 Java 8 吧。

說到 JVM 端設(shè)置，堆大小這個參數(shù)至關(guān)重要。雖然在后面我們還會討論如何調(diào)優(yōu) Kafka 性能的問題，但現(xiàn)在我想無腦給出一個通用的建議：將你的 JVM 堆大小設(shè)置成 6GB 吧，這是目前業(yè)界比較公認(rèn)的一個合理值。我見過很多人就是使用默認(rèn)的 Heap Size 來跑 Kafka，說實話默認(rèn)的 1GB 有點(diǎn)小，畢竟 Kafka Broker 在與客戶端進(jìn)行交互時會在 JVM 堆上創(chuàng)建大量的 ByteBuffer 實例，Heap Size 不能太小。

JVM 端配置的另一個重要參數(shù)就是垃圾回收器的設(shè)置，也就是平時常說的 GC 設(shè)置。如果你依然在使用 Java 7，那么可以根據(jù)以下法則選擇合適的垃圾回收器：

• 如果 Broker 所在機(jī)器的 CPU 資源非常充裕，建議使用 CMS 收集器。啟用方法是指定-XX:+UseCurrentMarkSweepGC。

• 否則，使用吞吐量收集器。開啟方法是指定-XX:+UseParallelGC。

當(dāng)然了，如果你已經(jīng)在使用 Java 9 了，那么就用默認(rèn)的 G1 收集器就好了。在沒有任何調(diào)優(yōu)的情況下，G1 表現(xiàn)得要比 CMS 出色，主要體現(xiàn)在更少的 Full GC，需要調(diào)整的參數(shù)更少等，所以使用 G1 就好了

現(xiàn)在我們確定好了要設(shè)置的 JVM 參數(shù)，我們該如何為 Kafka 進(jìn)行設(shè)置呢？有些奇怪的是，這個問題居然在 Kafka 官網(wǎng)沒有被提及。其實設(shè)置的方法也很簡單，你只需要設(shè)置下面這兩個環(huán)境變量即可：

• KAFKA_HEAP_OPTS：指定堆大小

• KAFKA_JVM_PERFORMANCE_OPTS：指定 GC 參數(shù)。

比如你可以這樣啟動 Kafka Broker，即在啟動 Kafka Broker 之前，先設(shè)置上這兩個環(huán)境變量：

$> export KAFKA_HEAP_OPTS=--Xms6g  --Xmx6g
$> export KAFKA_JVM_PERFORMANCE_OPTS= -server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -Djava.awt.headless=true
$> bin/kafka-server-start.sh config/server.properties

操作系統(tǒng)參數(shù)

最后我們來聊聊 Kafka 集群通常都需要設(shè)置哪些操作系統(tǒng)參數(shù)。通常情況下，Kafka 并不需要設(shè)置太多的 OS 參數(shù)，但有些因素最好還是關(guān)注一下，比如下面這幾個：

• 文件描述符限制
• 文件系統(tǒng)類型
• Swappiness
• 提交時間

首先是ulimit -n。我覺得任何一個 Java 項目最好都調(diào)整下這個值。實際上，文件描述符系統(tǒng)資源并不像我們想象的那樣昂貴，你不用太擔(dān)心調(diào)大此值會有什么不利的影響。通常情況下將它設(shè)置成一個超大的值是合理的做法，比如ulimit -n 1000000。其實設(shè)置這個參數(shù)一點(diǎn)都不重要，但不設(shè)置的話后果很嚴(yán)重，比如你會經(jīng)?？吹健癟oo many open files”的錯誤。

其次是文件系統(tǒng)類型的選擇。這里所說的文件系統(tǒng)指的是如 ext3、ext4 或 XFS 這樣的日志型文件系統(tǒng)。根據(jù)官網(wǎng)的測試報告，XFS 的性能要強(qiáng)于 ext4，所以生產(chǎn)環(huán)境最好還是使用 XFS。對了，最近有個 Kafka 使用 ZFS 的數(shù)據(jù)報告，貌似性能更加強(qiáng)勁，有條件的話不妨一試。

第三是 swap 的調(diào)優(yōu)。網(wǎng)上很多文章都提到設(shè)置其為 0，將 swap 完全禁掉以防止 Kafka 進(jìn)程使用 swap 空間。我個人反倒覺得還是不要設(shè)置成 0 比較好，我們可以設(shè)置成一個較小的值。為什么呢？因為一旦設(shè)置成 0，當(dāng)物理內(nèi)存耗盡時，操作系統(tǒng)會觸發(fā) OOM killer 這個組件，它會隨機(jī)挑選一個進(jìn)程然后 kill 掉，即根本不給用戶任何的預(yù)警。但如果設(shè)置成一個比較小的值，當(dāng)開始使用 swap 空間時，你至少能夠觀測到 Broker 性能開始出現(xiàn)急劇下降，從而給你進(jìn)一步調(diào)優(yōu)和診斷問題的時間。基于這個考慮，我個人建議將 swappniess 配置成一個接近 0 但不為 0 的值，比如 1。

最后是提交時間或者說是 Flush 落盤時間。向 Kafka 發(fā)送數(shù)據(jù)并不是真要等數(shù)據(jù)被寫入磁盤才會認(rèn)為成功，而是只要數(shù)據(jù)被寫入到操作系統(tǒng)的頁緩存（Page Cache）上就可以了，隨后操作系統(tǒng)根據(jù) LRU 算法會定期將頁緩存上的“臟”數(shù)據(jù)落盤到物理磁盤上。這個定期就是由提交時間來確定的，默認(rèn)是 5 秒。一般情況下我們會認(rèn)為這個時間太頻繁了，可以適當(dāng)?shù)卦黾犹峤婚g隔來降低物理磁盤的寫操作。當(dāng)然你可能會有這樣的疑問：如果在頁緩存中的數(shù)據(jù)在寫入到磁盤前機(jī)器宕機(jī)了，那豈不是數(shù)據(jù)就丟失了。的確，這種情況數(shù)據(jù)確實就丟失了，但鑒于 Kafka 在軟件層面已經(jīng)提供了多副本的冗余機(jī)制，因此這里稍微拉大提交間隔去換取性能還是一個合理的做法。