Quarkus流式输出故障处理：如何标记响应异常并关闭连接？

问题场景 #

运行Quarkus服务器传输大型数据集，处理过程中可能抛出异常。现有代码如下：

@GET
@Path("{fileName}/example")
@Produces(MediaType.APPLICATION_JSON)
fun example(@PathParam("fileName") fileName: String): Multi<Int> {
    return Multi.createFrom().iterable((0 .. 10)).map { if (it != 4) it else throw IllegalArgumentException() }
}

• 未处理异常时：接口输出[0,1,2,3后停止，连接保持挂起（curl请求会卡住）；
• 使用.onFailure().recoverWithCompletion()处理：流正常关闭，客户端得到[0,1,2,3]，但无法感知数据流发生了异常。

需求：既要关闭连接，又要让客户端知晓数据流处理异常（或保留畸形响应）。

解决方案 #

方案1：插入错误标记后关闭流（推荐） #

在异常发生时，向流中插入一个明确的错误标识，然后立即完成流。客户端收到包含错误标记的JSON后，可识别出异常状态，同时连接正常关闭。

@GET
@Path("{fileName}/example")
@Produces(MediaType.APPLICATION_JSON)
fun example(@PathParam("fileName") fileName: String): Multi<Any> {
    return Multi.createFrom().iterable((0 .. 10))
        .map { 
            if (it != 4) it else throw IllegalArgumentException("数据处理在位置$it出错") 
        }
        .onFailure()
            .recoverWithMulti { exception ->
                // 发送错误标识，随后立即完成流
                Multi.createFrom().items("ERROR: ${exception.message}")
                    .onItem().call { Multi.createFrom().completion() }
            }
}

客户端最终会收到类似[0,1,2,3,"ERROR: 数据处理在位置4出错"]的响应，既能确认流已结束，又能明确异常原因。

方案2：主动取消流保留畸形响应 #

如果需要严格保留截断的畸形JSON（比如[0,1,2,3），可在异常时主动取消流，强制关闭连接。客户端会收到截断的响应，需自行通过校验JSON完整性来判断异常。

@GET
@Path("{fileName}/example")
@Produces(MediaType.APPLICATION_JSON)
fun example(@PathParam("fileName") fileName: String): Multi<Int> {
    return Multi.createFrom().iterable((0 .. 10))
        .map { 
            if (it != 4) it else throw IllegalArgumentException() 
        }
        .onFailure()
            .invoke { 
                // 主动取消流，触发连接关闭
                it.cancel()
            }
}

注意事项 #

• 由于JSON流式输出是逐步发送的，一旦开始输出数组，无法回滚已发送的内容，也无法修改HTTP状态码（状态码已在响应头发送）；
• 方案1的错误标记需和客户端约定格式，确保客户端能正确识别；
• 方案2的畸形响应需要客户端具备错误处理逻辑（比如捕获JSON解析异常）。

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