案例分析

處理kafka consumer的程式的時候，發現如下錯誤：

ERROR [2017-01-12 07:16:02,466] com.flow.kafka.consumer.main.KafkaConsumer: Unexpected Error Occurred
! kafka.common.MessageSizeTooLargeException: Found a message larger than the maximum fetch size of this consumer on topic codeTopic partition 3 at fetch offset 94. Increase the fetch size, or decrease the maximum message size the broker will allow.
! at kafka.consumer.ConsumerIterator.makeNext(ConsumerIterator.scala:91) ~[pip-kafka-consumer.jar:na]
! at kafka.consumer.ConsumerIterator.makeNext(ConsumerIterator.scala:33) ~[pip-kafka-consumer.jar:na]
! at kafka.utils.IteratorTemplate.maybeComputeNext(IteratorTemplate.scala:66) ~[pip-kafka-consumer.jar:na]
! at kafka.utils.IteratorTemplate.hasNext(IteratorTemplate.scala:58) ~[pip-kafka-consumer.jar:na]
! at com.flow.kafka.consumer.main.KafkaConsumer$KafkaRiverFetcher.run(KafkaConsumer.java:291) ~[original-pip-kafka-consumer.jar:na]
! at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) [na:1.7.0_51]
! at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) [na:1.7.0_51]
! at java.lang.Thread.run(Thread.java:744) [na:1.7.0_51]
 

如上log可以看出，問題就是有一個較大的訊息資料在codeTopic的partition 3上，然後consumer未能消費，提示我可以減小broker允許進入的訊息資料的大小，或者增大consumer程式消費資料的大小。

從log上來看一目瞭然，如果要解決當前問題的話，

1. 減小broker訊息體大小(設定message.max.bytes引數)；
2. 增大consumer獲取資料資訊大小(設定fetch.message.max.bytes引數)。預設broker訊息體大小為1000000位元組即為1M大小。

消費者方面:fetch.message.max.bytes——>這將決定消費者可以獲取的資料大小。
broker方面:replica.fetch.max.bytes——>這將允許broker的副本傳送訊息在叢集並確保訊息被正確地複製。如果這是太小,則訊息不會被複制,因此,消費者永遠不會看到的訊息,因為訊息永遠不會承諾(完全複製)。
broker方面:message.max.bytes——>可以接受資料生產者最大訊息資料大小。

由我的場景來看較大的訊息體已經進入到了kafka，我這裡要解決這個問題，只需要增加consumer的fetch.message.max.bytes數值就好。我單獨把那條資料消費出來，寫到一個檔案中發現那條訊息大小為1.5M左右，為了避免再次發生這種問題我把consumer程式的fetch.message.max.bytes引數調節為了3072000即為3M，重啟consumer程式，檢視log一切正常，解決這個消費錯誤到此結束，下面介紹一下kafka針對大資料處理的思考。

kafka的設計初衷

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

針對這個問題，可以參考如下建議：

• 最好的方法是不直接傳送這些大的資料。如果有共享儲存，如NAS, HDFS, S3等，可以把這些大的檔案存放到共享儲存，然後使用Kafka來傳送檔案的位置資訊。

• 第二個方法是，將大的訊息資料切片或切塊，在生產端將資料切片為10K大小，使用分割槽主鍵確保一個大訊息的所有部分會被髮送到同一個kafka分割槽（這樣每一部分的拆分順序得以保留），如此以來，當消費端使用時會將這些部分重新還原為原始的訊息。

• 第三，Kafka的生產端可以壓縮訊息，如果原始訊息是XML，當通過壓縮之後，訊息可能會變得不那麼大。在生產端的配置引數中使用compression.codec和commpressed.topics可以開啟壓縮功能，壓縮演算法可以使用GZip或Snappy。

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

broker 配置

message.max.bytes (預設:1000000) – broker能接收訊息的最大位元組數，這個值應該比消費端的fetch.message.max.bytes更小才對，否則broker就會因為消費端無法使用這個訊息而掛起。
log.segment.bytes (預設: 1GB) – kafka資料檔案的大小，確保這個數值大於一個訊息的長度。一般說來使用預設值即可（一般一個訊息很難大於1G，因為這是一個訊息系統，而不是檔案系統）。
replica.fetch.max.bytes (預設: 1MB) – broker可複製的訊息的最大位元組數。這個值應該比message.max.bytes大，否則broker會接收此訊息，但無法將此訊息複製出去，從而造成資料丟失。

Consumer 配置

fetch.message.max.bytes (預設 1MB) – 消費者能讀取的最大訊息。這個值應該大於或等於message.max.bytes。所以，如果你一定要選擇kafka來傳送大的訊息，還有些事項需要考慮。要傳送大的訊息，不是當出現問題之後再來考慮如何解決，而是在一開始設計的時候，就要考慮到大訊息對叢集和主題的影響。

效能: 根據前面提到的效能測試，kafka在訊息為10K時吞吐量達到最大，更大的訊息會降低吞吐量，在設計叢集的容量時，尤其要考慮這點。
可用的記憶體和分割槽數：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引數為更大的超時時間。