Kafka制作人关闭后如何重新连接？

Question

我有一个多线程应用程序，它使用producer类生成消息，之前我使用下面的代码为每个请求创建producer。其中KafkaProducer是新建的，每个请求如下：

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

ProducerRecord<String, byte[]> data = new ProducerRecord<String, byte[]>(topic, objBytes);
producer.send(data, new Callback() {

                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception != null) {
                        isValidMsg[0] = false;
                        exception.printStackTrace();
                        saveOrUpdateLog(msgBean, producerType, exception);
                        logger.error("ERROR:Unable to produce message.",exception);
                    }
                }
            });
producer.close();

然后我阅读了关于生产者的Kafka文档，并了解到我们应该使用单个生产者实例来获得良好的性能。

然后我在一个singleton类中创建了KafkaProducer的单个实例。

现在什么时候

java.lang.IllegalStateException: Cannot send after the producer is closed.

或者我们如何在关闭后重新连接到生产者。问题是如果程序崩溃或有异常，那么？

Answer 1

Kafka producer应该是线程安全的，并且对它的线程池很节俭。你可能想用

producer.flush();

代替

producer.close();

让制作人保持开放状态，直到程序终止或您确定不再需要它。

如果仍要关闭生产者，请按需重新创建。

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

Answer 2

如javadoc for 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，请仅在使用完毕后将其关闭。如果您仍然希望确保消息在方法完成之前实际发送出去，而不是在缓冲区中等待，那么请使用< code > KafkaProducer # flush()，它将一直阻塞到当前缓冲区发送出去。如果您愿意，也可以在< code>Future#get()上阻塞。

如果你不打算关闭你的KafkaProducer，也需要注意一个警告（例如，在短期应用程序中，你只发送一些数据，应用程序在发送后立即终止）。KafkaProducer IO线程是一个守护进程线程，这意味着JVM不会等到该线程完成后才终止VM。因此，要确保您的消息确实被发送，请使用KafkaProducer#flush（），不要在Future#get。

Answer 3

通常，在 KafkaProducer 上调用 close（） 足以确保所有机上记录都已完成：

/**
 * Close this producer. This method blocks until all previously sent requests complete.
 * This method is equivalent to <code>close(Long.MAX_VALUE, TimeUnit.MILLISECONDS)</code>.
 * <p>
 * <strong>If close() is called from {@link Callback}, a warning message will be logged and close(0, TimeUnit.MILLISECONDS)
 * will be called instead. We do this because the sender thread would otherwise try to join itself and
 * block forever.</strong>
 * <p>
 *
 * @throws InterruptException If the thread is interrupted while blocked
 */

如果在应用程序的整个生命周期中都在使用生产者，那么在得到终止信号之前不要关闭它，然后调用< code>close()。如文档中所述，在多线程环境中使用生成器是安全的，因此您应该重用同一个实例。

如果您在多个线程中共享您的 Kafka 生产者，则有两种选择：

< li >在通过< code>Runtime.getRuntime()注册关闭回调时调用< code>close()。addShutdownHook来自您的主执行线程 < li >让您的多线程方法竞赛结束，只允许一个方法获胜。

2的粗略草图可能如下所示：

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
      }
    }
  }
}