Ktor接收Multipart时触发“Flow invariant is violated”异常问题求助

我是Ktor的新手，最近在实现服务器接收文件上传的功能时遇到了一个奇怪的异常。下面是我的路由代码：

fun Application.module() {
    install(ContentNegotiation) {
        json()
    }
    routing {
        post("/upload") {
            val multipart = call.receiveMultipart()
            multipart.forEachPart { part ->
                // 还没写具体处理逻辑
            }
            call.respond(HttpStatusCode.OK, "Test")
        }
    }
}

我用curl命令测试上传：

curl -X POST http://localhost:8080/upload -F "file=log_25_3.xml"

结果服务器抛出了这个异常：

java.lang.IllegalStateException: Flow invariant is violated: Flow was collected in [io.ktor.server.engine.DefaultUncaughtExceptionHandler@3777f889, CoroutineName(call-handler), io.ktor.server.netty.NettyDispatcher$CurrentContext@782e8f51, StandaloneCoroutine{Active}@4a837c5d, io.ktor.server.application.ClassLoaderAwareContinuationInterceptor@443a3274], but emission happened in [io.ktor.server.engine.DefaultUncaughtExceptionHandler@3777f889, CoroutineName(call-handler), io.ktor.server.netty.NettyDispatcher$CurrentContext@782e8f51, StandaloneCoroutine{Active}@4a837c5d, io.ktor.server.application.ClassLoaderAwareContinuationInterceptor@443a3274]. Please refer to 'flow' documentation or use 'flowOn' instead
at kotlinx.coroutines.flow.internal.SafeCollector_commonKt.checkContext(SafeCollector.common.kt:84)
at kotlinx.coroutines.flow.internal.SafeCollector.checkContext(SafeCollector.kt:132)
at kotlinx.coroutines.flow.internal.SafeCollector.emit(SafeCollector.kt:109)
at kotlinx.coroutines.flow.internal.SafeCollector.emit(SafeCollector.kt:82)
at io.ktor.http.content.MultipartKt$asFlow$1.invokeSuspend(Multipart.kt:145)
at io.ktor.http.content.MultipartKt$asFlow$1.invoke(Multipart.kt)
at io.ktor.http.content.MultipartKt$asFlow$1.invoke(Multipart.kt)
at kotlinx.coroutines.flow.SafeFlow.collectSafely(Builders.kt:57)

我本来以为这是处理文件上传的常规写法，但没想到会出现这个异常。有没有大佬知道这是怎么回事？

可能的解决方案和分析 #

这个异常看起来有点矛盾——报错说收集和发射的上下文不一致，但打印出来的上下文明明是完全一样的。这通常和Ktor的协程上下文处理、或者Multipart流的收集方式有关。这里提供几个可以尝试的解决方向：

• 确保正确处理Multipart部件的释放：即使你还没写处理逻辑，也需要调用part.dispose()来释放资源，或者用use块自动管理：

  multipart.forEachPart { part ->
      part.use {
          // 后续处理逻辑
      }
  }
  

  未正确释放部件可能会导致流的状态异常，进而触发上下文检查的错误。

• 显式指定流的上下文：虽然报错里上下文看起来一致，但可以尝试用flowOn切换到IO调度器，强制统一上下文：

  val multipart = call.receiveMultipart()
  multipart.flow.flowOn(Dispatchers.IO).collect { part ->
      part.use {
          // 处理部件
      }
  }
  

  （注意这里用flow.collect代替forEachPart，配合flowOn）

• 检查Ktor版本：这个问题可能是特定版本的Ktor的bug，建议你升级到最新的稳定版（比如2.x系列的最新版本），很多旧版本的协程上下文相关bug已经被修复。

• 避免在接收Multipart后立即respond：有时候在流还没完全收集完成就发送响应，也可能触发异常。确保所有部件处理完成后再调用call.respond，比如把respond放在collect块的最后：

  multipart.flow.flowOn(Dispatchers.IO).collect { part ->
      part.use {
          // 处理逻辑
      }
  }
  call.respond(HttpStatusCode.OK, "Test")
  

另外，你也可以尝试在application.conf里调整Netty的调度器配置，不过这个属于进阶操作，先尝试前面几个简单的方案。

内容来源于stack exchange