Ktor中Kafka Producer多线程优化：有序性与内存泄漏问题

问题场景 #

我有一个使用RSocket发送事件的Ktor应用，现需将这些事件同步发送至Kafka，初始代码如下：

suspend fun publishEvents(flow: SharedFlow<Pair<String, String>>) {
    val producer = KafkaProducer<String, String>(mapOf())
    flow.collect { 
        producer.send(ProducerRecord("topic", it.first, it.second))
    }
}

默认配置下单分区场景下，该Producer每秒无法处理超过5000条单条250字节的事件，调整分区数、压缩、linger、批量大小及acks等配置均无改善，似乎触及性能上限。

改为固定线程池后，性能提升但出现新问题：

suspend fun publishEvents(flow: SharedFlow<Pair<String, String>>) {
    val producer = KafkaProducer<String, String>(mapOf())
    val executorService = Executors.newFixedThreadPool(threads)
    flow.collect { 
        executorService.submit {
            producer.send(ProducerRecord("topic", it.first, it.second))
        }
    }
}

双线程下每秒可处理近10000条事件，但同一key的事件有序性无法保障，此外该方案存在内存泄漏问题，且瓶颈疑似与协程/SharedFlow相关。

需要解决：

• 同一key事件的有序性问题
• 内存泄漏问题
• Kafka Producer是否有线程数配置项？

解决方案 #

1. 保障同一Key的事件有序性 #

Kafka中同一Key的消息会被路由到同一分区，天然保证分区内有序，但多线程并发发送时，send的异步回调可能因网络延迟等原因导致后发先到，破坏顺序。解决方式：

• 按Key分组串行发送：将SharedFlow中的事件按Key分组，每组单独串行处理发送，结合协程Mutex保证串行性：

suspend fun publishEvents(flow: SharedFlow<Pair<String, String>>) {
    val producer = KafkaProducer<String, String>(getKafkaConfig())
    val keyLocks = mutableMapOf<String, Mutex>()
    try {
        flow.collect { (key, value) ->
            val mutex = keyLocks.getOrPut(key) { Mutex() }
            mutex.withLock {
                // 异步发送+回调确认，避免阻塞协程
                producer.send(ProducerRecord("topic", key, value)) { exception, _ ->
                    exception?.let { /* 处理发送失败逻辑 */ }
                }
            }
        }
    } finally {
        producer.close()
    }
}

这种方式既保留多Key并发的性能，又严格保证同一Key的消息顺序。

2. 修复内存泄漏问题 #

原线程池方案的内存泄漏源于：

• ExecutorService未关闭，线程长期存活
• KafkaProducer未正确关闭，持有连接等资源

修复方式：

• 用协程CoroutineScope管理线程池，配合asCoroutineDispatcher实现资源自动释放
• 确保在协程结束或取消时调用producer.close()

优化后的代码示例：

suspend fun publishEvents(flow: SharedFlow<Pair<String, String>>, threads: Int = 2) = coroutineScope {
    val producer = KafkaProducer<String, String>(getKafkaConfig())
    val dispatcher = Executors.newFixedThreadPool(threads).asCoroutineDispatcher()
    try {
        flow.collect { (key, value) ->
            launch(dispatcher) {
                // 若需保证Key有序，需结合上述Mutex方案
                producer.send(ProducerRecord("topic", key, value)) { exception, _ ->
                    exception?.let { /* 处理失败 */ }
                }
            }
        }
    } finally {
        dispatcher.close() // 释放线程池资源
        producer.close()   // 关闭KafkaProducer
    }
}

同时要确保SharedFlow的收集能响应协程取消，触发finally块执行关闭逻辑。

3. Kafka Producer的线程数配置 #

Kafka Producer本身不提供直接配置发送线程数的参数，因为内部已维护IO线程池处理异步发送任务。你之前遇到的单线程性能瓶颈，核心原因是在协程中同步阻塞调用send，浪费了协程的异步优势。

正确做法是利用Producer的异步特性，避免阻塞协程：

// 异步发送，不阻塞当前协程，由Producer内部IO线程处理
producer.send(ProducerRecord("topic", key, value)) { exception, recordMetadata ->
    if (exception != null) {
        // 处理发送失败
    }
}

这种方式无需额外创建线程池，既能利用Producer的异步性能，又避免多线程带来的有序性问题。

额外优化建议 #

• 检查SharedFlow的replay、extraBufferCapacity参数，避免因背压限制导致事件堆积，影响吞吐量
• 调整Producer的batch.size、linger.ms参数，让Producer尽可能批量发送消息，减少网络请求次数
• 避免在collect块中执行阻塞操作，优先使用协程异步API

内容的提问来源于stack exchange，提问作者AmsterdamLuis