Kotlin+Ktor视频流服务报Broken pipe，请求排查问题

核心问题分析 #

报错Broken pipe本质是客户端（浏览器）在接收响应前主动断开连接，根源在于响应头设置错误、响应内容不符合HTTP规范，以及内存预加载导致的响应延迟。

具体问题点与修复步骤 #

1. ContentType构造错误 #

原代码中ContentType的contentType和contentSubtype重复赋值为httpContentType（比如video/mp4），导致最终Content-Type格式错误，浏览器无法识别视频类型。

修复：
拆分httpContentType为类型和子类型：

val contentTypeParts = chunkWithMetadata.metadata.httpContentType.split("/")
val contentType = ContentType(contentTypeParts[0], contentTypeParts[1])

2. 未处理无Range头的请求 #

浏览器首次加载视频时通常不带Range头，此时应返回200 OK并传输完整文件，而非强制返回206 Partial Content，否则浏览器会拒绝处理响应。

修复：
在Handler中添加Range头判断：

val rangeHeader = call.request.headers["Range"]
if (rangeHeader == null) {
    // 返回完整文件
    call.response.header("Content-Type", metadata.httpContentType)
    call.response.header("Content-Length", metadata.size.toString())
    call.response.header("Accept-Ranges", "bytes")
    call.respondOutputStream(status = HttpStatusCode.OK, contentType = contentType) {
        storageService.getInputStream(uuid, 0, metadata.size).use { input ->
            input.copyTo(this)
        }
    }
    return@get
}

3. 预加载Chunk到内存导致的问题 #

原代码中readChunk用readAllBytes()将整个Chunk加载到ByteArray，大视频会引发内存溢出，且如果客户端中途断开连接，已加载的内存无法释放，同时延迟了响应发送时间，导致浏览器提前断开。

修复：
修改Service直接返回MinIO的InputStream，流式写入响应输出流：

// 修改Service方法
override suspend fun fetchChunk(uuid: UUID, range: Range): Pair<FileMetadataEntity, InputStream> {
    val fileMetadata = fileMetadataDAOFacade.findById(uuid) ?: throw FileNotFoundException()
    val start = range.getRangeStart()
    val chunkSize = range.getRangeEnd(fileMetadata.size) - start + 1
    val inputStream = storageService.getInputStream(uuid, start, chunkSize)
    return Pair(fileMetadata, inputStream)
}

// Handler中流式写入
call.respondOutputStream(
    status = HttpStatusCode.PartialContent,
    contentType = contentType,
    contentLength = chunkSize.toLong()
) {
    chunkInputStream.use { input ->
        input.copyTo(this)
    }
}

4. 响应头规范校验 #

确保Content-Range格式严格符合HTTP规范：bytes start-end/total，Content-Length必须等于end - start + 1的数值，与实际传输的字节数一致。

修复示例：

val start = parsedRange.getRangeStart()
val end = parsedRange.getRangeEnd(metadata.size)
val contentRange = "bytes $start-$end/${metadata.size}"
val contentLength = (end - start + 1).toString()

call.response.header("Content-Range", contentRange)
call.response.header("Content-Length", contentLength)

5. 空值与异常处理 #

原代码中直接用fileMetadata!!强制解包，若文件不存在会抛出NPE，应捕获并返回404 Not Found；同时处理MinIO读取异常，返回500 Internal Server Error。

修复：

val fileMetadata = fileMetadataDAOFacade.findById(uuid) ?: run {
    call.respond(HttpStatusCode.NotFound)
    return@get
}

完整修复后的Handler示例 #

get("/{uuid}") {
    val uuidStr = call.parameters["uuid"] ?: run {
        call.respond(HttpStatusCode.BadRequest)
        return@get
    }
    val uuid = try {
        UUID.fromString(uuidStr)
    } catch (e: IllegalArgumentException) {
        call.respond(HttpStatusCode.BadRequest)
        return@get
    }

    val metadata = fileMetadataDAOFacade.findById(uuid) ?: run {
        call.respond(HttpStatusCode.NotFound)
        return@get
    }

    val contentTypeParts = metadata.httpContentType.split("/")
    val contentType = ContentType(contentTypeParts[0], contentTypeParts[1])

    val rangeHeader = call.request.headers["Range"]
    if (rangeHeader == null) {
        // 返回完整文件
        call.response.header("Content-Type", metadata.httpContentType)
        call.response.header("Content-Length", metadata.size.toString())
        call.response.header("Accept-Ranges", "bytes")
        call.respondOutputStream(status = HttpStatusCode.OK, contentType = contentType) {
            runCatching {
                storageService.getInputStream(uuid, 0, metadata.size).use { input ->
                    input.copyTo(this)
                }
            }.onFailure {
                call.respond(HttpStatusCode.InternalServerError)
            }
        }
        return@get
    }

    // 处理Range请求
    val parsedRange = Range.parseHttpRangeString(rangeHeader, defaultChunkSize)
    val start = parsedRange.getRangeStart()
    val end = parsedRange.getRangeEnd(metadata.size)
    val chunkSize = end - start + 1

    call.response.header("Content-Type", metadata.httpContentType)
    call.response.header("Accept-Ranges", "bytes")
    call.response.header("Content-Length", chunkSize.toString())
    call.response.header("Content-Range", "bytes $start-$end/${metadata.size}")

    call.respondOutputStream(
        status = HttpStatusCode.PartialContent,
        contentType = contentType,
        contentLength = chunkSize.toLong()
    ) {
        runCatching {
            storageService.getInputStream(uuid, start, chunkSize).use { input ->
                input.copyTo(this)
            }
        }.onFailure {
            // 忽略Broken pipe异常，这是客户端主动断开的正常情况
            if (it !is IOException || it.message != "Broken pipe") {
                call.respond(HttpStatusCode.InternalServerError)
            }
        }
    }
}

关键注意事项 #

• 流式传输避免内存溢出：始终直接操作InputStream，不要预加载大文件到内存。
• 严格遵循HTTP规范：200和206响应的头信息必须符合标准，浏览器对视频流的头信息校验非常严格。
• 忽略Broken pipe异常：当客户端主动断开连接时，该异常属于正常情况，无需返回错误响应。

内容的提问来源于stack exchange，提问作者Александр