本篇文章介绍Kafka处理大文件出现内存溢出 java.lang.OutOfMemoryError: Direct buffer memory,主要内容包括基础应用、实用技巧、原理机制等方面,希望对大家有所帮助。
kafka是由scala和java编写的。因此须要调一些jvm的参数。java的内存分为堆内内存和堆外内存。
-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的方式,而这个申请的是堆外内存。
Kafka设计的初衷是迅速处理短小的消息,一般10K大小的消息吞吐性能最好。但有时候,我们需要处理更大的消息,比如XML文档或JSON内容,一个消息差不多有10-100M,这种情况下,Kakfa应该如何处理?
不过如果上述方法都不是你需要的,而你最终还是希望传送大的消息,那么,则可以在kafka中设置下面一些参数:
注意:message.max.bytes,要设置大于发送最大数据的大小,不然会produce失败。
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,不然消费会失败。
receive.message.max.bytes
只支持1000到1000000000。否则会出现 “Receive failed: Invalid message size 1047207987 (0..1000000000): increase receive.message.max.bytes”这样的错误。
注意:一定要选择kafka来传送大的消息,还有些事项需要考虑。要传送大的消息,不是当出现问题之后再来考虑如何解决,而是在一开始设计的时候,就要考虑到大消息对集群和主题的影响。
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参数为更大的超时时间。
G1>CMS的优势
G1适合的场景:
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磁盘。
在写文件的时候,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的缓冲设置为原来的两倍。
手机扫一扫
移动阅读更方便
你可能感兴趣的文章