1

我有一个应用程序应该向 Kafka 发送有限数量的消息然后退出。出于某种原因，即使我关闭了生产者，Kafka 连接也会保持正常。我的实现（在 Scala 中）或多或少

object Kafka {
  private val props = new Properties()

  props.put("compression.codec", DefaultCompressionCodec.codec.toString)
  props.put("producer.type", "sync")
  props.put("metadata.broker.list", "localhost:9092")
  props.put("batch.num.messages", "200")
  props.put("message.send.max.retries", "3")
  props.put("request.required.acks", "-1")
  props.put("client.id", "myclient")

  private val producer = new Producer[Array[Byte], Array[Byte]](new ProducerConfig(props))
  private def encode(msg: Message) = new KeyedMessage("topic", msg.id.getBytes, write(msg).getBytes)

  def send(msg: Message) = Try(producer.send(encode(msg)))
  def close() = producer.close()
}

这Message是一个简单的案例类，我如何将其转换为字节数组并不重要。

消息确实到达了，但是当我最终调用时Kafka.close()，应用程序没有退出，并且连接似乎没有被释放。

有没有办法明确要求 Kafka 终止连接？

4

1 回答 1

1

def close() = 生产者.close()

这将创建一个名为“close”的函数，该函数调用 producer.close() 我没有看到任何证据表明您的代码实际上关闭了生产者。

你只需要调用：producer.close

于 2014-11-06T14:49:47.593 回答