java - 卡夫卡生产者巨大的内存使用(泄漏？)

Question

我们在生产环境中部署了 3 个代理 Kafka 0.10.1.0。有些应用程序嵌入了 Kafka Producer，它们将应用程序日志发送到某个主题。该主题有 10 个分区，复制因子为 3。

我们观察到，其中一些应用程序服务器上的内存使用量间歇性地不断飙升。在进行堆转储后，我们发现最可疑的是:

**org.apache.kafka.common.network.Selector -**

occupies 352,519,104 (24.96%) bytes. The memory is accumulated in one instance of "byte[]" loaded by "<system class loader>".

**org.apache.kafka.common.network.KafkaChannel -**

occupies 352,527,424 (24.96%) bytes. The memory is accumulated in one instance of "byte[]" loaded by "<system class loader>"

<小时/>

两者都拥有大约 352MB 的空间。 3 个这样的实例，因此它们消耗了大约 1.2GB 的内存。

现在关于生产者的使用。没有大量日志被发送到 Kafka 集群。大约为 200 条消息/秒。整个应用程序中仅使用一个生产者对象。使用异步发送功能。

造成如此巨大内存使用的原因是什么？在这个特定的 Kafka 版本中，这是某种内存泄漏吗？

这是生产中使用的 Kafka Producer 配置。

kafka.bootstrap.servers=x.x.x.x:9092,x.x.x.x:9092,x.x.x.x:9092
kafka.acks=0
kafka.key.serializer=org.apache.kafka.common.serialization.StringSerializer
kafka.value.serializer=org.apache.kafka.common.serialization.StringSerializer
kafka.max.block.ms=1000
kafka.request.timeout.ms=1000
kafka.max.in.flight.requests.per.connection=1
kafka.retries=0
kafka.compression.type=gzip
kafka.security.protocol=SSL
kafka.ssl.truststore.location=/data/kafka/kafka-server-truststore.jks
kafka.ssl.truststore.password=XXXXXX
kafka.linger.ms=300
logger.level=INFO

这是 GC 日志中显示 Kafka 网络线程分配的部分

<allocation-stats totalBytes="3636833992" >
  <allocated-bytes non-tlh="3525405200" tlh="111428792" />
  <largest-consumer threadName="kafka-producer-network-thread | producer-1" threadId="0000000033A26700" bytes="3525287448" />
</allocation-stats>
<gc-op id="591417" type="scavenge" timems="21.255" contextid="591414" timestamp="2018-09-19T17:55:32.938">
  <scavenger-info tenureage="14" tenuremask="4000" tiltratio="89" />
  <memory-copied type="nursery" objects="61155" bytes="6304384" bytesdiscarded="3968416" />
  <memory-copied type="tenure" objects="1199" bytes="230312" bytesdiscarded="38656" />
  <finalization candidates="461" enqueued="316" />
  <ownableSynchronizers candidates="18" cleared="5" />
  <references type="soft" candidates="231" cleared="0" enqueued="0" dynamicThreshold="23" maxThreshold="32" />
  <references type="weak" candidates="20" cleared="2" enqueued="1" />
  <references type="phantom" candidates="2" cleared="0" enqueued="0" />
</gc-op>
<gc-end id="591418" type="scavenge" contextid="591414" durationms="21.715" usertimems="11.640" systemtimems="0.125" timestamp="2018-09-19T17:55:32.939" activeThreads="64">
  <mem-info id="591419" free="4226106664" total="6049234944" percent="69">
    <mem type="nursery" free="3855164752" total="4294967296" percent="89">
      <mem type="allocate" free="3855164752" total="3865444352" percent="99" />
      <mem type="survivor" free="0" total="429522944" percent="0" />
    </mem>
    <mem type="tenure" free="370941912" total="1754267648" percent="21">
      <mem type="soa" free="362646600" total="1740233728" percent="20" />
      <mem type="loa" free="8295312" total="14033920" percent="59" />
    </mem>
    <pending-finalizers system="315" default="1" reference="1" classloader="0" />
    <remembered-set count="4110" />
  </mem-info>
</gc-end>
<cycle-end id="591420" type="scavenge" contextid="591414" timestamp="2018-09-19T17:55:32.940" />
<allocation-satisfied id="591421" threadId="0000000033A26700" bytesRequested="352518920" />
<af-end id="591422" timestamp="2018-09-19T17:55:32.962" />
<exclusive-end id="591423" timestamp="2018-09-19T17:55:32.962" durationms="45.987" />

Answer 1

原因可能有很多。但如果您需要优化，可以尝试以下操作:

1. replica.fetch.max.bytes-每个分区的缓冲区大小。分区数乘以最大消息大小不超过可用内存。同样适用于消费者 - fetch.message.max.bytes、max.partition.fetch.bytes - 服务器将返回的每个分区的最大数据量。可以通过使用replica.high.watermark.checkpoint.interval.ms来完成丢失数据的检查点，这将调整吞吐量。

2.batch.size(不应超过可用内存)和linger.ms(设置缓冲数据的最长时间，如果是异步的)