Spring Boot+Ktor批量下载文件时遇Chunked流异常求助

问题背景 #

基于Kotlin 1.8.10 + Spring Boot 3.0.4搭建的应用，通过Spring Integration实现的RabbitMQ消费者从第三方服务器下载文件。小批量（数百条消息）处理完全正常，但在数万条消息的初始导入场景下，运行一段时间后会抛出EOF异常：

java.io.EOFException: Chunked stream has ended unexpectedly: no chunk size
at io.ktor.http.cio.ChunkedTransferEncodingKt.decodeChunked(ChunkedTransferEncoding.kt:76)
...

用curl单独请求异常对应的资源能正常获取响应，断点调试ChunkedTransferEncoding.kt:76时，发现input（ByteBufferChannel）的_state为terminated（正常状态应为writing）。

可能原因排查 #

• 连接池资源耗尽：高并发下载场景下，Ktor客户端连接池被占满，新请求无法获取连接，或已有连接因空闲超时被提前关闭。
• 第三方服务器限流：短时间内大量请求触发服务器限流策略，服务器主动断开连接，导致Chunked传输未完成就终止（单次curl请求不受限流影响，但批量请求会触发）。
• Ktor客户端配置缺失：未设置合理的超时时间（连接/读取超时），导致请求中途被客户端主动关闭；未启用重试机制，临时断开无法自动恢复。
• 消费者并发过高：RabbitMQ消费者并发数设置过大，同时发起的下载请求远超服务器或自身系统承载能力，引发连接不稳定。

修复方案 #

1. 优化Ktor客户端配置 #

调整连接池、超时和重试参数，示例配置如下：

val httpClient = HttpClient(CIO) {
    // 配置各类超时
    install(HttpTimeout) {
        requestTimeoutMillis = 60_000   // 整体请求超时60秒
        connectTimeoutMillis = 10_000  // 连接建立超时10秒
        socketTimeoutMillis = 30_000    // 数据读取超时30秒
    }
    // 配置连接池
    engine {
        maxConnectionsCount = 50        // 全局最大并发连接数
        endpoint {
            maxConnectionsPerRoute = 20 // 单域名路由最大连接数
            keepAliveTime = 30_000      // 连接空闲保持时间
            connectTimeout = 10_000
            socketTimeout = 30_000
        }
    }
    // 启用异常重试机制
    install(Retry) {
        retryOnExceptionIf(3) { _, cause ->
            // 针对IO异常、服务器异常重试3次
            cause is IOException || cause is ServerResponseException
        }
        delayMillis { retry -> retry * 1000 } // 指数退避延迟
    }
}

2. 调整RabbitMQ消费者并发度 #

降低消费者并发数，避免短时间内发起过多请求：

@Bean
public IntegrationFlow rabbitMqInboundFlow(ConnectionFactory connectionFactory) {
    return IntegrationFlows.from(Amqp.inboundAdapter(connectionFactory, "download-queue")
                    .concurrentConsumers(5)    // 基础并发数，建议5-10
                    .maxConcurrentConsumers(10)) // 最大并发数
            .handle(message -> {
                // 执行文件下载逻辑
            })
            .get();
}

3. 增加失败补偿机制 #

在下载逻辑中捕获IO异常，将失败消息重新放回队列（依赖RabbitMQ的重试机制）：

fun downloadFile(url: String): File? {
    return try {
        httpClient.get(url).readBytes().let { bytes ->
            File("downloads/${UUID.randomUUID()}").apply { writeBytes(bytes) }
        }
    } catch (e: EOFException) {
        log.error("文件下载中断，URL: $url", e)
        // 抛出异常触发RabbitMQ重试
        throw e
    } catch (e: Exception) {
        log.error("文件下载失败，URL: $url", e)
        null
    }
}

4. 排查第三方服务器限流规则 #

联系第三方服务器运维，确认是否存在QPS、连接数等限流限制，根据对方要求调整请求频率（比如添加固定延迟）。

5. 启用Ktor调试日志 #

开启Ktor客户端DEBUG级日志，追踪连接建立、传输阶段的详细过程，定位连接终止的具体时机：

# application.properties
logging.level.io.ktor=DEBUG
logging.level.io.ktor.http.cio=DEBUG

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