处理亿级记录CSV时Executor Service内存不足的解决方法咨询

你的代码出现OOM和运行停滞的核心原因是任务生产速度远大于消费速度：

• forEachLine会快速读取CSV的每一行，立刻将HTTP请求任务提交到固定大小的线程池（10个线程）。当HTTP请求的处理速度跟不上CSV读取速度时，线程池的任务队列会无限累积未执行的任务，最终耗尽内存。
• 额外问题：HTTP请求未关闭响应导致连接泄漏，连接池耗尽后所有请求阻塞，进一步加剧任务队列积压；连接池未配置合理大小，无法支撑并发请求。

1. 控制任务提交速率，避免队列无限膨胀 #

使用有界队列的线程池，配合Semaphore（信号量）控制同时提交的任务数，确保生产速度和消费速度匹配。

2. 正确处理HTTP响应，避免连接泄漏 #

httpClient.execute(post)返回的CloseableHttpResponse必须关闭，同时要消费响应实体，否则连接池会被耗尽，导致后续请求阻塞。

3. 合理配置HttpClient连接池 #

设置总最大连接数、每个路由的最大连接数，适配线程池的并发规模。

4. 优化CSV解析（可选但重要） #

直接用split(",")无法处理带引号、包含逗号的CSV字段，建议使用成熟的CSV解析库（如OpenCSV）避免数据解析错误。

import org.apache.http.client.methods.CloseableHttpResponse
import org.apache.http.client.methods.HttpPost
import org.apache.http.entity.StringEntity
import org.apache.http.impl.client.CloseableHttpClient
import org.apache.http.impl.client.HttpClients
import org.apache.http.impl.conn.PoolingHttpClientConnectionManager
import org.apache.http.util.EntityUtils
import java.util.concurrent.Executors
import java.util.concurrent.Semaphore
import java.util.concurrent.TimeUnit

class CsvProcessor {
    // 线程池大小和信号量许可数保持一致，控制并发数
    private val nThreads = 10
    // 使用有界队列，避免任务无限积压
    private val executorService = Executors.newFixedThreadPool(nThreads)
    // 信号量控制同时提交的任务数，和线程池大小匹配
    private val semaphore = Semaphore(nThreads)

    private val manager = PoolingHttpClientConnectionManager().apply {
        // 设置总最大连接数
        maxTotal = nThreads * 2
        // 设置每个路由的最大连接数
        defaultMaxPerRoute = nThreads
    }

    private val httpClient: CloseableHttpClient = HttpClients.custom()
        .setConnectionManager(manager)
        .build()

    fun send(body: String) {
        var response: CloseableHttpResponse? = null
        try {
            val post = HttpPost("url")
            post.setHeader("Content-Type", "application/json")
            post.entity = StringEntity(body)
            response = httpClient.execute(post)
            // 必须消费响应实体，否则连接可能无法复用
            EntityUtils.consume(response.entity)
        } catch (e: Exception) {
            // 这里添加异常处理逻辑，比如重试、日志记录
            e.printStackTrace()
        } finally {
            response?.close()
        }
    }

    fun process() {
        var lineCount = 0
        ResourceUtils.getFile("classpath:file.csv").bufferedReader().forEachLine { line ->
            lineCount++
            if (lineCount > 1) { // 跳过表头
                // 获取信号量许可，没有许可时会阻塞，直到有任务完成释放许可
                semaphore.acquire()
                val records = line.split(",") // 建议替换为OpenCSV解析
                executorService.execute {
                    try {
                        send(generate(records[0], records[1], records[2], records[3]))
                    } finally {
                        // 任务完成后释放信号量许可
                        semaphore.release()
                    }
                }
                if (lineCount % 10000 == 0) {
                    println("已处理 $lineCount 条记录")
                }
            }
        }

        executorService.shutdown()
        try {
            if (!executorService.awaitTermination(1, TimeUnit.HOURS)) { // 设置合理的超时时间，避免无限等待
                executorService.shutdownNow()
            }
        } catch (e: InterruptedException) {
            System.err.println(e.message)
            executorService.shutdownNow()
            Thread.currentThread().interrupt()
        }
    }

    private fun generate(id: String, firstName: String, lastName: String, dateTime: String): String {
        return """{"id":"$id","firstName":"$firstName","lastName":"$lastName","dateTime":"$dateTime"}"""
    }
}

• 批量处理：如果目标服务支持批量接口，可以将多条CSV记录打包成一个JSON数组发送，减少HTTP请求次数，提升整体效率。
• 内存监控：添加内存使用监控日志，比如定期打印JVM堆内存使用情况，方便排查问题。
• 异常重试：对失败的HTTP请求添加重试机制，避免个别失败请求导致数据丢失。

内容的提问来源于stack exchange，提问作者J.Doe