我从多个生产者向kafka发送数据时得到InterruptedException

Question

我试图运行kafka生产者使用kafka-clients API。我有多个生产者与单独的线程运行，每个试图写数据到Kafka。问题是，当我增加并行运行的线程数量时，我会从kafka中得到一个中断的异常。例如，如果我并行运行20个线程，它不会引发任何异常，但当我并行运行100个线程时，我会得到以下异常：

线程“pool-910-thread-1”org.apache.kafka.common.errors.interruptexception:java.lang.interruptedexception
在org.apache.kafka.clients.producer.kafkaProducer.close(kafkaProducer.java:1154)
在org.apache.kafka.clients.producer.kafkaProducer.close(kafkaProducer.java:1154)
在Processor.runProducer(IEC104KafKareadMessageProcessor.java:45)
在com.t4e.iec104.connection.iec60870ReadListener.writeToJsonFile(IEC60870ReadListener.java:707)
在com.t4e.iec104.connection.iec60870ReadListener.newasdu(IEC60870ReadListener.java:75)
在org.openmuc.j60870.connection util.concurrent.ThreadPoolExecutor$worker.run(threadPoolExecutor.java:624)
在java.lang.thread.run(thread.java:748)
导致了y:java.lang.interruptedException
在java.lang.object.wait（本机方法）
在java.lang.thread.join(thread.java:1260)
在org.apache.kafka.clients.producer.kafkaProducer.close(kafkaProducer.java:1152)

下面是我的制作者代码：

private static final Logger logger = LoggerFactory.getLogger(IEC104KafkaReadMessageProcessor.class);
    static KafkaProducerConfigReader kafkaConfig = new KafkaProducerConfigReader();
    static String newLine = System.getProperty("line.separator");

    /**
     * @param message
     * @return
     * @throws InterruptedException
     * @throws ExecutionException
     */
    public static synchronized RecordMetadata runProducer(String message) throws InterruptedException, ExecutionException {
        Producer<Long, String> producer = ProducerCreator.createProducer();
        ProducerRecord<Long, String> record = new ProducerRecord<Long, String>(kafkaConfig.getTopicName(), message);
        try {
            RecordMetadata metadata = producer.send(record).get();
            logger.info(("Record sent with key " + " to partition " + metadata.partition() + " with offset "
                    + metadata.offset()));
            return metadata;
        } catch (ExecutionException e) {
            logger.error("ExecutionException : Error in sending record to kafka");
            throw new ExecutionException(e);
        } catch (InterruptedException e) {
            logger.error("InterruptedException : Error in sending record" + newLine);
            throw new InterruptedException();
        } finally {
            logger.info(" Closing Kafka producer ");
            producer.close();
        }
    }

Answer 1

问题可能不在您正在创建生产者的这段代码中。看着你们分享的日志，

线程“pool-910-thread-1”org.apache.kafka.common.errors.interruptexception：java.lang.interruptedexception

我可以看到有910个线程池处于活动状态。创建这么多的池而不是创建线程保存一个池可能是一个更好的主意。您可能希望查看创建线程池的位置并对其进行控制。

我怀疑您代码中的线程泄漏会导致此中断异常。