kafka使用者抛出java.lang.outofmemoryerror:直接缓冲区内存

krcsximq  于 2021-06-07  发布在  Kafka
关注(0)|答案(1)|浏览(598)

我使用的是单节点kafka代理（0.10.2）和单节点zookeeper代理（3.4.9）。我有一个消费者服务器（单核和1.5 gb内存）。每当我运行一个有5个或更多线程的进程时，在抛出这些异常之后，我的使用者的线程就会被杀死
例外情况1
java.lang.outofmemoryerror:java.nio.heapbytebuffer处的java堆空间。java:57)在java.nio.bytebuffer.allocate（bytebuffer。java:335)在org.apache.kafka.common.network.networkreceive.readfromreadablechannel（networkreceive。java:93)在org.apache.kafka.common.network.networkreceive.readfrom（networkreceive。java:71)在org.apache.kafka.common.network.kafkachannel.receive（kafkachannel。java:169)在org.apache.kafka.common.network.kafkachannel.read（kafkachannel。java:150)在org.apache.kafka.common.network.selector.pollselectionkeys（selector。java:355)在org.apache.kafka.common.network.selector.poll（selector。java:303)在org.apache.kafka.clients.networkclient.poll（networkclient。java:349)在org.apache.kafka.clients.consumer.internals.consumernetworkclient.poll（consumernetworkclient。java:226)在org.apache.kafka.clients.consumer.internals.consumernetworkclient.pollnowakup（consumernetworkclient。java:263)在org.apache.kafka.clients.consumer.internals.abstractcoordinator$heartbeatthread.run（abstractcoordinator。java:887)
例外情况2
kafka协调器心跳线程中未捕获的异常| topic1:java.lang.outofmemoryerror:java.nio.bits.reservememory（bits）处的直接缓冲区内存。java:693)在java.nio.directbytebuffer。java:123)位于java.nio.bytebuffer.allocatedirect（bytebuffer。java:311)在sun.nio.ch.util.gettemporarydirectbuffer（util。java:241)在sun.nio.ch.ioutil.read（ioutil。java:195)在sun.nio.ch.socketchannelimpl.read（socketchannelimpl。java:380)在org.apache.kafka.common.network.plaintexttransportlayer.read（plaintexttransportlayer。java:110)在org.apache.kafka.common.network.networkreceive.readfromreadablechannel（networkreceive。java:97)在org.apache.kafka.common.network.networkreceive.readfrom（networkreceive。java:71)在org.apache.kafka.common.network.kafkachannel.receive（kafkachannel。java:169)在org.apache.kafka.common.network.kafkachannel.read（kafkachannel。java:150)在org.apache.kafka.common.network.selector.pollselectionkeys（selector。java:355)在org.apache.kafka.common.network.selector.poll（选择器。java:303)在org.apache.kafka.clients.networkclient.poll（networkclient。java:349)在org.apache.kafka.clients.consumer.internals.consumernetworkclient.poll（consumernetworkclient。java:226)在org.apache.kafka.clients.consumer.internals.consumernetworkclient.pollnowakup（consumernetworkclient。java:263)在org.apache.kafka.clients.consumer.internals.abstractcoordinator$heartbeatthread.run（abstractcoordinator。java:887)
我在google上搜索并使用了下面提到的jvm参数，但仍然出现了相同的异常
-xx:maxdirectmemorysize=768m
-xms512m型
如何解决这个问题？是否需要任何其他javm参数调整？
我的Kafka消费代码是

  1. import com.mongodb.DBObject
  2. import org.apache.kafka.clients.consumer.ConsumerRebalanceListener
  3. import org.apache.kafka.clients.consumer.ConsumerRecord
  4. import org.apache.kafka.clients.consumer.ConsumerRecords
  5. import org.apache.kafka.clients.consumer.KafkaConsumer
  6. import org.apache.kafka.clients.consumer.OffsetAndMetadata
  7. import org.apache.kafka.clients.consumer.OffsetCommitCallback
  8. import org.apache.kafka.common.TopicPartition
  9. import org.apache.kafka.common.errors.InterruptException
  10. import org.apache.kafka.common.errors.WakeupException
  11. import org.slf4j.Logger
  12. import org.slf4j.LoggerFactory
  13. import java.util.regex.Pattern
  15. class KafkaPollingConsumer implements Runnable {
  16. private static final Logger logger = LoggerFactory.getLogger(KafkaPollingConsumer.class)
  17. private static final String TAG = "[KafkaPollingConsumer]"
  18. private final KafkaConsumer<String, byte []> kafkaConsumer
  19. private Map<TopicPartition,OffsetAndMetadata> currentOffsetsMap = new HashMap<>()
  20. List topicNameList
  21. Map kafkaTopicConfigMap = new HashMap<String,Object>()
  22. Map kafkaTopicMessageListMap = new HashMap<String,List>()
  23. Boolean isRebalancingTriggered = false
  24. private final Long REBALANCING_SLEEP_TIME = 1000
  26. public KafkaPollingConsumer(String serverType, String groupName, String topicNameRegex, Integer batchSize, Integer maxPollTime, Integer requestTime){
  27.     logger.debug("{} [Constructor] [Enter] Thread Name {} serverType group Name TopicNameRegex",TAG,Thread.currentThread().getName(),serverType,groupName,topicNameRegex)
  28.     logger.debug("Populating Property for kafak consumer")
  29.     logger.debug("BatchSize {}",batchSize)
  30.     Properties kafkaConsumerProperties = new Properties()
  31.     kafkaConsumerProperties.put("group.id", groupName)
  32.     kafkaConsumerProperties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
  33.     kafkaConsumerProperties.put("value.deserializer", "com.custom.kafkaconsumerv2.deserializer.CustomObjectDeserializer")
  34.     switch(serverType){
  35.         case KafkaTopicConfigEntity.KAFKA_NODE_TYPE_ENUM.Priority.toString() :
  36.             kafkaConsumerProperties.put("bootstrap.servers",ConfigLoader.conf.kafkaServer.priority.kafkaNode)
  37.             kafkaConsumerProperties.put("enable.auto.commit",ConfigLoader.conf.kafkaServer.priority.consumer.enable.auto.commit)
  38.             kafkaConsumerProperties.put("auto.offset.reset",ConfigLoader.conf.kafkaServer.priority.consumer.auto.offset.reset)
  39.             break
  40.         case KafkaTopicConfigEntity.KAFKA_NODE_TYPE_ENUM.Bulk.toString() :
  41.             kafkaConsumerProperties.put("bootstrap.servers",ConfigLoader.conf.kafkaServer.bulk.kafkaNode)
  42.             kafkaConsumerProperties.put("enable.auto.commit",ConfigLoader.conf.kafkaServer.bulk.consumer.enable.auto.commit)
  43.             kafkaConsumerProperties.put("auto.offset.reset",ConfigLoader.conf.kafkaServer.bulk.consumer.auto.offset.reset)
  44.             kafkaConsumerProperties.put("max.poll.records",1)
  45.             kafkaConsumerProperties.put("max.poll.interval.ms",600000)
  46.             kafkaConsumerProperties.put("request.timeout.ms",600005)
  47.             break
  48.         default :
  49.             throw "Invalid server type"
  50.             break
  51.     }
  52.     logger.debug("{} [Constructor] KafkaConsumer Property Populated {}",properties.toString())
  53.     kafkaConsumer = new KafkaConsumer<String, byte []>(kafkaConsumerProperties)
  54.     topicNameList = topicNameRegex.split(Pattern.quote('|'))
  55.     logger.debug("{} [Constructor] Kafkatopic List {}",topicNameList.toString())
  56.     logger.debug("{} [Constructor] Exit",TAG)
  57. }
  59. private class HandleRebalance implements ConsumerRebalanceListener {
  60.     public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
  61.         logger.error('{} In onPartitionAssigned setting isRebalancingTriggered to false',TAG)
  62.         isRebalancingTriggered = false
  63.     }
  65.     public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
  66.         logger.error("{} In onPartitionsRevoked setting osRebalancingTriggered to true",TAG)
  67.         isRebalancingTriggered = true
  68.         publishAllKafkaTopicBatchMessages()
  69.         commitOffset()
  71.     }
  72. }
  74. private class AsyncCommitCallBack implements OffsetCommitCallback{
  76.     @Override
  77.     void onComplete(Map<TopicPartition, OffsetAndMetadata> map, Exception e) {
  79.     }
  80. }
  82. @Override
  83. void run() {
  84.     logger.debug("{} Starting Thread ThreadName {}",TAG,Thread.currentThread().getName())
  85.     populateKafkaConfigMap()
  86.     initializeKafkaTopicMessageListMap()
  87.     String topicName
  88.     String consumerClassName
  89.     String consumerMethodName
  90.     Boolean isBatchJob
  91.     Integer batchSize = 0
  92.     final Thread mainThread = Thread.currentThread()
  93.     Runtime.getRuntime().addShutdownHook(new Thread() {
  94.         public void run() {
  95.             logger.error("{},gracefully shutdowning thread {}",TAG,mainThread.getName())
  96.             kafkaConsumer.wakeup()
  97.             try {
  98.                 mainThread.join()
  99.             } catch (InterruptedException exception) {
  100.                 logger.error("{} Error : {}",TAG,exception.getStackTrace().join("\n"))
  101.             }
  102.         }
  103.     })
  104.     kafkaConsumer.subscribe(topicNameList , new HandleRebalance())
  105.     try{
  106.         while(true){
  107.             logger.debug("{} Starting Consumer with polling time in ms 100",TAG)
  108.             ConsumerRecords kafkaRecords
  109.             if(isRebalancingTriggered == false) {
  110.                 kafkaRecords = kafkaConsumer.poll(100)
  111.             }
  112.             else{
  113.                 logger.error("{} in rebalancing going to sleep",TAG)
  114.                 Thread.sleep(REBALANCING_SLEEP_TIME)
  115.                 continue
  116.             }
  117.             for(ConsumerRecord record: kafkaRecords){
  118.                 if(isRebalancingTriggered == true){
  119.                     break
  120.                 }
  121.                 currentOffsetsMap.put(new TopicPartition(record.topic(), record.partition()),new OffsetAndMetadata(record.offset() +1))
  122.                 topicName = record.topic()
  123.                 DBObject kafkaTopicConfigDBObject = kafkaTopicConfigMap.get(topicName)
  124.                 consumerClassName = kafkaTopicConfigDBObject.get(KafkaTopicConfigEntity.CLASS_NAME_KEY)
  125.                 consumerMethodName = kafkaTopicConfigDBObject.get(KafkaTopicConfigEntity.METHOD_NAME_KEY)
  126.                 isBatchJob = kafkaTopicConfigDBObject.get(KafkaTopicConfigEntity.IS_BATCH_JOB_KEY)
  127.                 logger.debug("Details about Message")
  128.                 logger.debug("Thread {}",mainThread.getName())
  129.                 logger.debug("Topic {}",topicName)
  130.                 logger.debug("Partition {}",record.partition().toString())
  131.                 logger.debug("Offset {}",record.offset().toString())
  132.                 logger.debug("clasName {}",consumerClassName)
  133.                 logger.debug("methodName {}",consumerMethodName)
  134.                 logger.debug("isBatchJob {}",isBatchJob.toString())
  135.                 Object message = record.value()
  136.                 logger.debug("message {}",message.toString())
  137.                 if(isBatchJob == true){
  138.                     prepareMessagesBatch(topicName,message)
  139.                     //batchSize = Integer.parseInt(kafkaTopicConfigDBObject.get(KafkaTopicConfigEntity.BATCH_SIZE_KEY).toString())
  140.                     //logger.debug("batchSize {}",batchSize.toString())
  141.                 }
  142.                 else{
  143.                     publishMessageToNonBatchConsumer(consumerClassName,consumerMethodName,message)
  144.                 }
  145.                 //publishMessageToConsumers(consumerClassName,consumerMethodName,isBatchJob,batchSize,message,topicName)
  146.                 //try {
  147.                 //  kafkaConsumer.commitAsync(currentOffsetsMap,new AsyncCommitCallBack())
  148.                 logger.debug("{} Commiting Messages to Kafka",TAG)
  149.                 //}
  150.                 /*catch(Exception exception){
  151.                     kafkaConsumer.commitSync(currentOffsetsMap)
  152.                     currentOffsetsMap.clear()
  153.                     logger.error("{} Error while commiting async so commiting in sync {}",TAG,exception.getStackTrace().join("\n"))
  154.                 }*/
  155.             }
  156.             commitOffset()
  157.             publishAllKafkaTopicBatchMessages()
  158.         }
  159.     }
  160.     catch(InterruptException exception){
  161.         logger.error("{} In InterruptException",TAG)
  162.         logger.error("{} In Exception exception message {}",TAG,exception.getMessage())
  163.         logger.error("{} Exception {}",TAG,exception.getStackTrace().join("\n"))
  164.     }
  165.     catch (WakeupException exception) {
  166.         logger.error("{} In WakeUp Exception",TAG)
  167.         logger.error("{} In Exception exception message {}",TAG,exception.getMessage())
  168.         logger.error("{} Exception {}",TAG,exception.getStackTrace().join("\n"))
  169.     }
  170.     catch(Exception exception){
  171.         exception.getMessage()
  172.         logger.error("{} In Exception",TAG)
  173.         logger.error("{} In Exception exception message {}",TAG,exception.getMessage())
  174.         logger.error("{} Exception {}",TAG,exception.getStackTrace().join("\n"))
  175.     }
  176.     finally {
  177.         logger.error("{} In finally commiting remaining offset ",TAG)
  178.         publishAllKafkaTopicBatchMessages()
  179.         //kafkaConsumer.commitSync(currentOffsetsMap)
  180.         kafkaConsumer.close()
  181.         logger.error("{} Exiting Consumer",TAG)
  182.     }
  183. }
  185. private void commitOffset(){
  186.     logger.debug("{} [commitOffset] Enter")
  187.     logger.debug("{} currentOffsetMap {}",currentOffsetsMap.toString())
  188.     if(currentOffsetsMap.size() > 0) {
  189.         kafkaConsumer.commitSync(currentOffsetsMap)
  190.         currentOffsetsMap.clear()
  191.     }
  192.     logger.debug("{} [commitOffset] Exit")
  194. }
  196. private void publishMessageToConsumers(String consumerClassName,String consumerMethodName,Boolean isBatchJob,Integer batchSize,Object message, String topicName){
  197.     logger.debug("{} [publishMessageToConsumer] Enter",TAG)
  198.     if(isBatchJob == true){
  199.         publishMessageToBatchConsumer(consumerClassName, consumerMethodName,batchSize, message, topicName)
  200.     }
  201.     else{
  202.         publishMessageToNonBatchConsumer(consumerClassName, consumerMethodName, message)
  203.     }
  204.     logger.debug("{} [publishMessageToConsumer] Exit",TAG)
  205. }
  207. private void publishMessageToNonBatchConsumer(String consumerClassName, String consumerMethodName, message){
  208.     logger.debug("{} [publishMessageToNonBatchConsumer] Enter",TAG)
  209.     executeConsumerMethod(consumerClassName,consumerMethodName,message)
  210.     logger.debug("{} [publishMessageToNonBatchConsumer] Exit",TAG)
  211. }
  213. private void publishMessageToBatchConsumer(String consumerClassName, String consumerMethodName, Integer batchSize, Object message, String topicName){
  214.     logger.debug("{} [publishMessageToBatchConsumer] Enter",TAG)
  215.     List consumerMessageList = kafkaTopicMessageListMap.get(topicName)
  216.     consumerMessageList.add(message)
  217.     if(consumerMessageList.size() == batchSize){
  218.         logger.debug("{} [publishMessageToBatchConsumer] Pushing Messages In Batches",TAG)
  219.         executeConsumerMethod(consumerClassName, consumerMethodName, consumerMessageList)
  220.         consumerMessageList.clear()
  221.     }
  222.     kafkaTopicMessageListMap.put(topicName,consumerMessageList)
  223.     logger.debug("{} [publishMessageToBatchConsumer] Exit",TAG)
  224. }
  226. private void populateKafkaConfigMap(){
  227.     logger.debug("{} [populateKafkaConfigMap] Enter",TAG)
  228.     KafkaTopicConfigDBService kafkaTopicConfigDBService = KafkaTopicConfigDBService.getInstance()
  229.     topicNameList.each { topicName ->
  230.         DBObject kafkaTopicDBObject = kafkaTopicConfigDBService.findByTopicName(topicName)
  231.         kafkaTopicConfigMap.put(topicName,kafkaTopicDBObject)
  232.     }
  233.     logger.debug("{} [populateKafkaConfigMap] kafkaConfigMap {}",TAG,kafkaTopicConfigMap.toString())
  234.     logger.debug("{} [populateKafkaConfigMap] Exit",TAG)
  235. }
  237. private void initializeKafkaTopicMessageListMap(){
  238.     logger.debug("{} [initializeKafkaTopicMessageListMap] Enter",TAG)
  239.     topicNameList.each { topicName ->
  240.         kafkaTopicMessageListMap.put(topicName,[])
  241.     }
  242.     logger.debug("{} [populateKafkaConfigMap] kafkaTopicMessageListMap {}",TAG,kafkaTopicMessageListMap.toString())
  243.     logger.debug("{} [initializeKafkaTopicMessageListMap] Exit",TAG)
  244. }
  246. private void executeConsumerMethod(String className, String methodName, def messages){
  247.     try{
  248.         logger.debug("{} [executeConsumerMethod] Enter",TAG)
  249.         logger.debug("{} [executeConsumerMethod] className  {} methodName {} messages {}",TAG,className,methodName,messages.toString())
  250.         Class.forName(className)."$methodName"(messages)
  251.     } catch (Exception exception){
  252.         logger.error("{} [{}] Error while executing method : {} of class: {} with params : {} - {}", TAG, Thread.currentThread().getName(), methodName,
  253.                 className, messages.toString(), exception.getStackTrace().join("\n"))
  254.     }
  255.     logger.debug("{} [executeConsumerMethod] Exit",TAG)
  256. }
  258. private void publishAllKafkaTopicBatchMessages(){
  259.     logger.debug("{} [publishAllKafkaTopicBatchMessages] Enter",TAG)
  260.     String consumerClassName = null
  261.     String consumerMethodName = null
  262.     kafkaTopicMessageListMap.each { topicName, messageList ->
  263.         if (messageList != null && messageList.size() > 0) {
  264.             DBObject kafkaTopicDBObject = kafkaTopicConfigMap.get(topicName)
  265.             consumerClassName = kafkaTopicDBObject.get(KafkaTopicConfigEntity.CLASS_NAME_KEY)
  266.             consumerMethodName = kafkaTopicDBObject.get(KafkaTopicConfigEntity.METHOD_NAME_KEY)
  267.             logger.debug("{} Pushing message in topic {} className {} methodName {} ", TAG, topicName, consumerClassName, consumerMethodName)
  268.             if (messageList != null && messageList.size() > 0) {
  269.                 executeConsumerMethod(consumerClassName, consumerMethodName, messageList)
  270.                 messageList.clear()
  271.                 kafkaTopicMessageListMap.put(topicName, messageList)
  273.             }
  274.         }
  275.     }
  276.     logger.debug("{} [publishAllKafkaTopicBatchMessages] Exit",TAG)
  277. }
  279. private void prepareMessagesBatch(String topicName,Object message){
  280.     logger.debug("{} [prepareMessagesBatch] Enter",TAG)
  281.     logger.debug("{} [prepareMessagesBatch] preparing batch for topic {}",TAG,topicName)
  282.     logger.debug("{} [prepareMessagesBatch] preparting batch for message {}",TAG,message.toString())
  283.     List consumerMessageList = kafkaTopicMessageListMap.get(topicName)
  284.     consumerMessageList.add(message)
  285.     kafkaTopicMessageListMap.put(topicName,consumerMessageList)
  287. }

}

Javaapache-kafkamultithreadingkafka-consumer-apijvm-arguments

1条答案

按热度按时间
ddhy6vgd

ddhy6vgd1#

kafka consumers通过以下两个参数处理数据积压，
最大轮询间隔毫秒
使用使用者群组管理时，调用poll（）之间的最大延迟。这为消费者在获取更多记录之前可以空闲的时间量设置了一个上限。如果在此超时过期之前未调用poll（），则认为使用者失败，组将重新平衡，以便将分区重新分配给另一个成员。
默认值为300000。
最大轮询记录数
对poll（）的单个调用中返回的最大记录数。
默认值为500。
忽略根据需求设置上述两个参数可能会导致轮询最大数据，而使用者可能无法处理可用资源，从而导致内存不足或有时无法提交使用者偏移量。因此，建议始终使用max.poll.records和max.poll.interval.ms参数。
在代码中，kafkatopicconfigentity.kafka\u node\u type\u enum.priority.tostring（）缺少这两个参数，这可能是导致轮询期间outofmemory问题的原因。

赞(0）分享  回复(0）举报  2021-06-07
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com