Kafka Producer线程，即使没有发送消息，也有大量线程

Question

我目前分析了我的kafka制作人Spring boot应用程序，发现许多“kafka制作人-网络-线程”正在运行（总共47个）。即使没有发送数据，它也永远不会停止运行。我的应用程序看起来有点像这样：

var kafkaSender = KafkaSender(kafkaTemplate, applicationProperties)
kafkaSender.sendToKafka(json, rs.getString("KEY"))

与KafkaSender：

@Service
class KafkaSender(val kafkaTemplate: KafkaTemplate<String, String>, val applicationProperties: ApplicationProperties) {

@Transactional(transactionManager = "kafkaTransactionManager")
fun sendToKafka(message: String, stringKey: String) {
   kafkaTemplate.executeInTransaction { kt ->
       kt.send(applicationProperties.kafka.topic, System.currentTimeMillis().mod(10).toInt(), System.currentTimeMillis().rem(10).toString(),
               message)
   }
}

companion object {
    val log = LoggerFactory.getLogger(KafkaSender::class.java)!!
}
}

因为每次我想给kafka发送消息时，我都会引用一个新的KafkaSender，所以我认为会创建一个新的线程，然后将消息发送到Kafka队列。目前看起来像是产生了一批生产者，但从来没有清理过，即使他们没有任何事情可做。

这种行为是有意的吗？

在我看来，这种行为应该与数据源池几乎相同，使线程保持活动状态一段时间，但是当无事可做时，请清除它。

Answer 1

使用事务时，生产者缓存会按需增长，而不会减少。

如果您正在侦听器容器（使用者）线程上生成消息；每个主题/分区/消费者组都有一个生产者。这是解决僵尸Geofence问题所必需的，以便在发生重新平衡且分区移动到其他实例时，事务id将保持不变，以便代理可以正确处理这种情况。

如果您不关心僵尸防护问题(并且您可以处理重复交付)，请将< code > DefaultKafkaProducerFactory 上的< code > producerPerConsumerPartition 属性设置为false，生产者的数量将会少得多。

编辑

从版本2.8开始，默认EOSMode现在是V2（又名BETA）；这意味着不再需要每个主题/分区/组都有一个生产者，只要代理版本是2.5或更高版本。