kafka处理大文件注意事项 - 简书

安宇雨 - 随手采集
2022-01-07 16:34:14
随手采集
0000-未整理-等待研究

这篇文章是我在粗略学习kafka时遇到的问题和网上查阅过的资料。kafka和java我都不是很熟悉，只知道基本的概念。文章中可能有疏漏和错误的地方，大家参考一下就好。有错误拜托大家指正！

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的内存分为堆内内存和堆外内存。

-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的什么组件的哪些方法用了堆内内存还是堆外内存，还有用多少，我不清楚，没有学过java，也没看过kafka源码。只是碰到了这个问题，然后记录一下。

producer端（php-rdkafka）

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

consumer端（php-rdkafka）

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。而consumer就会收到“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 参数。关于此参数的一些讨论：
https://stackoverflow.com/questions/39730126/difference-between-session-timeout-ms-and-max-poll-interval-ms-for-kafka-0-10-0
1. 当开启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