kafka producer应该是线程安全的，并且使用它的线程池非常节省。你可能想用

producer.flush();

而不是

producer.close();

在节目结束或你确定不再需要它之前，让制作人一直开放。
如果仍要关闭生产者，请按需重新创建。

producer = new KafkaProducer<String, byte[]>(prop);

如javadoc中所述 KafkaProducer :

public void close()

Close this producer. This method blocks until all previously sent requests complete.
This method is equivalent to close(Long.MAX_VALUE, TimeUnit.MILLISECONDS).

src公司：https://kafka.apache.org/0110/javadoc/org/apache/kafka/clients/producer/kafkaproducer.html#close()
因此，您不必担心您的消息不会被发送，即使您在发送后立即拨打close。
如果您计划多次使用kafkaproducer，请仅在使用完毕后关闭它。如果您仍然希望保证消息在方法完成之前实际发送，而不是在缓冲区中等待，那么使用 KafkaProducer#flush() 在发送当前缓冲区之前将被阻止。你也可以阻止 Future#get() 如果你愿意的话。
如果你不打算关闭kafkaproducer，还有一个要注意的问题（例如，在短期应用程序中，你只发送一些数据，发送后应用程序立即终止）。kafkaproducer io线程是守护进程线程，这意味着jvm不会等到该线程完成后才终止vm。因此，要确保您的消息真正被发送，请使用 KafkaProducer#flush() ，无参数 KafkaProducer#close() 或阻止 Future#get() .

这个 KafkaProducer.send 方法是异步的，它返回 Future[RecordMetadata] . 如果你 close 在发送之后，立即就有了竞争条件，并且由于 KafkaProducer 你的消息可能永远不会被发送。
如果您的生产者在应用程序的整个生命周期中都在使用，请不要关闭它，一旦应用程序终止，就让它死掉。如文档中所述，生产者在多线程环境中使用是安全的，因此您应该重用同一示例。
如果你仍然认为你需要关闭 KafkaProducer 在某些情况下，可以添加 isClosed 在kafka对象中标记它，并在使用者需要重新发送数据时监视它。草图可以是：

object KafkaOwner {
  private var producer: KafkaProducer = ???
  @volatile private var isClosed = false

  def close(): Unit = {
    if (!isClosed) {
      kafkaProducer.close()
      isClosed = true
    }
  }

  def instance: KafkaProducer = {
    this.synchronized {
      if (!isClosed) producer 
      else {
        producer = new KafkaProducer()
        isClosed = false
      }
    }
  }
}