主旨内容

本篇文章介绍Kafka处理大文件出现内存溢出 java.lang.OutOfMemoryError: Direct buffer memory，主要内容包括基础应用、实用技巧、原理机制等方面，希望对大家有所帮助。

broker端

配置文件须要配置的参数

• message.max.bytes : kafka会接收单个消息size的最大限制， 默认为1M左右。若是producer发送比这个大的消息，kafka默认会丢掉。producer能够从callback函数中得到错误码：10。

• log.segment.bytes : kafka数据文件的大小。默认为1G, 须要确保此值大于一个消息的最大大小。

• replica.fetch.max.bytes : broker可复制的消息的最大字节数, 默认为1M。这个值比message.max.bytes大，不然broker会接收此消息，但没法将此消息复制出去，从而形成数据丢失。

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端

message.max.bytes，要设置大于发送最大数据的大小，不然会produce失败。

consumer端

receive.message.max.bytes : kafka 协议response 的最大长度，应该保证次参数大于等于message.max.bytes。不然消费会失败。

版本太低的librdkafka的receive.message.max.bytes只支持1000到1000000000。最新版本的能够支持到2147483647。

• 使用此参数的时候还须要注意一个问题，在broker端设置的message.max.bytes为1000，consumer端设置的receive.message.max.bytes也为1000，可是除了数据，response还有协议相关字段，这时候整个response的大小就会超过1000。

“Receive failed: Invalid message size 1047207987 (0..1000000000): increase receive.message.max.bytes”这样的错误。

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).

自动调整到message.max.bytes这样的大小，返回的数据大小甚至还可能超过这个大小。

当设置receive.message.max.bytes == message.max.bytes == 1000 就会出上面说的那个错误。因此应该让consumer端设置的receive.message.max.bytes大于broker端设置的 message.max.bytes ，我猜应该大于单个最大数据的大小，这样就不会出错了，可是没有验证。

目前个人测试环境数据最大为500M左右。message.max.bytes 我设置了900M， receive.message.max.bytes设置了1000000000， 暂时没有出现问题。

内存方面须要考虑的问题

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 参数。

参考资料：

kafka的一些概念

• http://f.dataguru.cn/thread-730782-1-1.html
• https://www.cnblogs.com/huxi2b/p/6223228.html

kafka jvm 配置

• https://blog.csdn.net/chen88358323/article/details/51824232
• https://blog.csdn.net/xzj9581/article/details/72866225

kafka 内存考虑

https://www.cloudera.com/documentation/kafka/latest/topics/kafka_performance.html#concept_exp_hzk_br__d22306e79

• http://www.cnblogs.com/doubletree/p/4264969.html
• http://www.voidcn.com/article/p-zmwswtfy-bbz.html

kafka的复制

https://blog.csdn.net/lizhitao/article/details/51718185

kafka和DirectMemory的关系

http://www.360doc.com/content/13/0502/23/7669533_282552666.shtml

java的nio和内存管理介绍

• http://www.importnew.com/17781.html
• https://blog.csdn.net/flyineagle/article/details/1399554
• https://blog.csdn.net/evan_man/article/details/509105
• http://www.importnew.com/21463.html42
• http://blog.chinaunix.net/uid-26863299-id-3559878.html
• https://blog.csdn.net/u010003835/article/details/52957904

kafka出现的问题

• https://blog.csdn.net/guoyuqi0554/article/details/48630907
• http://blog.51cto.com/navyaijm/1931962

性能调优

• https://www.cnblogs.com/SpeakSoftlyLove/p/6511547.html
• https://blog.csdn.net/vegetable_bird_001/article/details/51858915session

参数讨论：

• https://stackoverflow.com/questions/39730126/difference-between-session-timeout-ms-and-max-poll-interval-ms-for-kafka-0-10-0

当开启auto commit时，当前一次poll获得消息对应的offset实际上是下一次调用poll时再提交的。即便过了auto.commit.interval.ms也不会提交。

• https://stackoverflow.com/questions/38230862/need-clarification-about-kafka-auto-commit-and-auto-commit-interval-ms