Kafka技術專題之「效能調優篇」訊息佇列伺服器端出現記憶體溢位OOM以及相關效能調優實戰分析

記憶體問題

本篇文章介紹Kafka處理大檔案出現記憶體溢位 java.lang.OutOfMemoryError: Direct buffer memory，主要內容包括基礎應用、實用技巧、原理機制等方面，希望對大家有所幫助。

bin目錄下的kafka-run-class.sh中須要設定的引數

kafka是由scala和java編寫的。因此須要調一些jvm的引數。java的記憶體分為堆內記憶體和堆外記憶體。

JVM引數系列

• -Xms2048m, -Xmx2048m，設定的是堆內記憶體。

• -Xms是初始可用的最大堆內記憶體。-Xmx設定的是最大可用的堆內記憶體。兩者設定成同樣是由於效率問題，可讓jvm少作一些運算。若是這兩個引數設定的過小，kafka會出現java.lang.OutOfMemoryError: Java heap space的錯誤。

• -XX:MaxDirectMemorySize=8192m。這個引數設定的過小，kafka會出現java.lang.OutOfMemoryError: Direct buffer memory的錯誤。 由於kafka的網路IO使用了java的nio中的DirectMemory的方式,而這個申請的是堆外記憶體。

producer端

Kafka設計的初衷是迅速處理短小的訊息，一般10K大小的訊息吞吐效能最好。但有時候，我們需要處理更大的訊息，比如XML檔案或JSON內容，一個訊息差不多有10-100M，這種情況下，Kakfa應該如何處理？

設定建議：

1. （傳輸資原始檔的位置-再消費的時候進行獲取真正的資源）首選的方法是不直接傳送這些大的資料。如果有共用儲存，如NAS, HDFS, S3等，可以把這些大的檔案存放到共用儲存，然後使用Kafka來傳送檔案的位置資訊。
2. （傳輸資原始檔的進行資料流的拆分，分批次傳送）將大的訊息資料切片或切塊，在生產端將資料切片為10K大小，使用分割區主鍵確保一個大訊息的所有部分會被傳送到同一個kafka分割區（這樣每一部分的拆分順序得以保留），如此以來，當消費端使用時會將這些部分重新還原為原始的訊息。
3. （傳輸資原始檔的進行資料流壓縮，降低傳輸量）Kafka的生產端可以壓縮訊息，如果原始訊息是XML，當通過壓縮之後，訊息可能會變得不那麼大。在生產端的設定引數中使用compression.codec和commpressed.topics可以開啟壓縮功能，壓縮演演算法可以使用GZip或Snappy。

不過如果上述方法都不是你需要的，而你最終還是希望傳送大的訊息，那麼，則可以在kafka中設定下面一些引數：

broker端

組態檔須要設定的引數

• message.max.bytes (預設:1000000) : kafka會接收單個訊息size的最大限制， 預設為1M左右。若是producer傳送比這個大的訊息，kafka預設會丟掉。producer能夠從callback函數中得到錯誤碼：10。設定了之後控制了broker能接收訊息的最大位元組數，這個值應該比消費端的fetch.message.max.bytes更小才對，否則broker就會因為消費端無法使用這個訊息而掛起。
• log.segment.bytes(預設: 1GB) : kafka資料檔案的大小。預設為1G, 須要確保此值大於一個訊息的最大大小。一般說來使用預設值即可（一般一個訊息很難大於1G，因為這是一個訊息系統，而不是檔案系統）。
• replica.fetch.max.bytes (預設: 1MB) : broker可複製的訊息的最大位元組數, 預設為1M。這個值比message.max.bytes大，不然broker會接收此訊息，但沒法將此訊息複製出去，從而形成資料丟失。

注意：message.max.bytes，要設定大於傳送最巨量資料的大小，不然會produce失敗。

consumer端

broker為什麼會返回總量為1000大小的資料呢？

librdkafka有這樣一個引數：fetch.max.bytes, 它有這樣的描述：

Maximum amount of data the broker shall return for a Fetch request. Messages are fetched in batches by the consumer and if the first message batch in the first non-empty partition of the Fetch request is larger than this value, then the message batch will still be returned to ensure the consumer can make progress. The maximum message batch size accepted by the broker is defined via message.max.bytes (broker config) or max.message.bytes (broker topic config). fetch.max.bytes is automatically adjusted upwards to be at least message.max.bytes (consumer config).

設定的引數

receive.message.max.bytes(預設 1MB) : kafka協定response的最大長度，也是消費者能讀取的最大訊息。這個值應該大於或等於message.max.bytes，不然消費會失敗。

• 較低版本的librdkafka的receive.message.max.bytes只支援1000到1000000000。
• 最新版本的能夠支援到2147483647。

否則會出現 「Receive failed: Invalid message size 1047207987 (0..1000000000): increase receive.message.max.bytes」這樣的錯誤。

• 使用此引數的時候還須要注意一個問題，在broker端設定的message.max.bytes為1000，consumer端設定的receive.message.max.bytes也為1000，可是除了資料，response還有協定相關欄位，這時候整個response的大小就會超過1000。

注意：一定要選擇kafka來傳送大的訊息，還有些事項需要考慮。要傳送大的訊息，不是當出現問題之後再來考慮如何解決，而是在一開始設計的時候，就要考慮到大訊息對叢集和主題的影響。

效能影響

• 根據前面提到的效能測試，kafka在訊息為10K時吞吐量達到最大，更大的訊息會降低吞吐量，在設計叢集的容量時，尤其要考慮這點。

可用的記憶體和分割區數造成OOM的場景

• Brokers會為每個分割區分配replica.fetch.max.bytes引數指定的記憶體空間，假設replica.fetch.max.bytes=1M，且有1000個分割區，則需要差不多1G的記憶體，確保分割區數 * 最大的訊息不會超過伺服器的記憶體，否則會報OOM錯誤。

• 消費端的fetch.message.max.bytes指定了最大訊息需要的記憶體空間，同樣，分割區數 * 最大需要記憶體空間不能超過伺服器的記憶體。所以，如果你有大的訊息要傳送，則在記憶體一定的情況下，只能使用較少的分割區數或者使用更大記憶體的伺服器。

垃圾回收

到現在為止，我在kafka的使用中還沒發現過此問題，但這應該是一個需要考慮的潛在問題。更大的訊息會讓GC的時間更長（因為broker需要分配更大的塊），隨時關注GC的紀錄檔和伺服器的紀錄檔資訊。如果長時間的GC導致kafka丟失了zookeeper的對談，則需要設定zookeeper.session.timeout.ms引數為更大的超時時間。

推薦使用1.7出來的G1垃圾回收機制代替CMS。

• G1>CMS的優勢

  • G1在壓縮空間方面有優勢
  • G1通過將記憶體空間分成區域（Region）的方式避免記憶體碎片問題
  • Eden, Survivor, Old區不再固定、在記憶體使用效率上來說更靈活
  • G1可以通過設定預期停頓時間（Pause Time）來控制垃圾收集時間避免應用雪崩現象
  • G1在回收記憶體後會馬上同時做合併空閒記憶體的工作、而CMS預設是在STW（stop the world）的時候做
  • G1會在Young GC中使用、而CMS只能在O區使用

• G1適合的場景：

  • 伺服器端多核CPU、JVM記憶體佔用較大的應用（至少大於4G）
  • 應用在執行過程中會產生大量記憶體碎片、需要經常壓縮空間
  • 想要更可控、可預期的GC停頓週期；防止高並行下應用雪崩現象
  • 我們的kafka的kafka-run-class.sh 中已經包含了

KAFKA_JVM_PERFORMANCE_OPTS="-server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+DisableExplicitGC -Djava.awt.headless=true"

所以只需要修改kafka-server-start.sh。這裡面將記憶體設定為4G，因為當前kafka的堆記憶體使用了800多M，1個G的記憶體不夠用。但是分配太多，也沒什麼用，還容易影響到pagecache，降低效率：

export KAFKA_HEAP_OPTS="-Xms4g -Xmx4g"

記憶體方面須要考慮的問題

Brokers allocate a buffer the size of replica.fetch.max.bytes for each partition they replicate. If replica.fetch.max.bytes is set to 1 MiB, and you have 1000 partitions, about 1 GiB of RAM is required. Ensure that the number of partitions multiplied by the size of the largest message does not exceed available memory.

The same consideration applies for the consumer fetch.message.max.bytes setting. Ensure that you have enough memory for the largest message for each partition the consumer replicates. With larger messages, you might need to use fewer partitions or provide more RAM.

若是一個訊息須要的處理時間很長，broker會認為consumer已經掛了，把partition分配給其餘的consumer，而後迴圈往復， 這條record永遠無法消費。方法是增長max.poll.interval.ms 引數。

提高效能調優能力

Don't fear the filesystem!中提到kafka使用page cache進行檔案儲存。計算機的記憶體分為虛擬記憶體和實體記憶體。實體記憶體是真實的記憶體，虛擬記憶體是用磁碟來代替記憶體。並通過swap機制實現磁碟到實體記憶體的載入和替換,這裡面用到的磁碟我們稱為swap磁碟。

pageCache機制

在寫檔案的時候，Linux首先將資料寫入沒有被使用的記憶體中，這些記憶體被叫做記憶體頁（page cache）。然後讀的時候，Linux會優先從page cache中查詢，如果找不到就會從硬碟中查詢。

當實體記憶體使用達到一定的比例後，Linux就會使用進行swap，使用磁碟作為虛擬記憶體。通過cat /proc/sys/vm/swappiness可以看到swap引數。這個參數列示虛擬記憶體中swap磁碟佔了多少百分比。0表示最大限度的使用記憶體，100表示儘量使用swap磁碟。

系統預設的引數是60，當實體記憶體使用率達到40%，就會頻繁進行swap，影響系統效能，推薦將vm.swappiness 設定為較低的值1。最終我設定為10，因為我們的機器的記憶體還是比較小的，只有40G，設定的太小，可能會影響到虛擬記憶體的使用吧。

髒檔案

當大量的持續不斷的資料寫入cache記憶體中後，這些資料就被稱為髒資料。需要儘快將這些髒資料flush到磁碟中，釋放記憶體。

這裡需要關注兩個引數：

• vm.dirty_background_ratio:這個引數指定了當檔案系統快取髒頁數量達到系統記憶體百分之多少時（如5%）就會觸發pdflush/flush/kdmflush等後臺回寫程序執行，將一定快取的髒頁非同步地刷入外存；

• vm.dirty_ratio:這個引數則指定了當檔案系統快取髒頁數量達到系統記憶體百分之多少時（如10%），系統不得不開始處理快取髒頁（因為此時髒頁數量已經比較多，為了避免資料丟失需要將一定髒頁刷入外存）；在此過程中很多應用程序可能會因為系統轉而處理檔案IO而阻塞。

這裡推薦將vm.dirty_background_ratio設定為5， vm.dirty_ratio有的人設定為10，但是我覺得太小了，還是預設的就可以了。

網路

kafka叢集對網路的要求比較高，可以將socket的緩衝設定為原來的兩倍。

• net.core.wmem_default 設定為128K
• net.core.rmem_default 設定為128K

學習資料

• 效能調優kafka和官方網站