如何在ExpediaGroup GraphQL中实现Reactor Flux并解决序列化错误

核心原因 #

报错的直接问题是GraphQL Schema中asynchronouslyDoFluxInt的类型定义错误：你将该字段定义为单个Int类型，但实际返回的是Flux<Int>（对应多元素集合，最终会被收集为ArrayList），类型不匹配导致序列化失败。

修正方案 #

1. 修正GraphQL Schema #

确保asynchronouslyDoFluxInt的类型为列表类型，示例如下：

type Query {
  asynchronouslyDo: Int
  asynchronouslyDoFluxInt: [Int] # 用方括号明确标记为列表类型
}

2. 正确的DataFetcher实现 #

你第二次尝试的CustomFunctionDataFetcher逻辑是正确的，还可以用Reactor自带方法简化代码：

class CustomFunctionDataFetcher(target: Any?, fn: KFunction<*>) : FunctionDataFetcher(target, fn) {
    override fun get(environment: DataFetchingEnvironment): Any? {
        val result = super.get(environment)
        return when (result) {
            is Mono<*> -> result.toFuture()
            is Flux<*> -> result.collectList().toFuture() // 替代collect(Collectors.toList())，代码更简洁
            else -> result
        }
    }
}

3. 为什么第一个版本的DataFetcher会返回null？ #

第一个版本对Flux的处理逻辑错误：

result.map { it -> Mono.just(it).toFuture() }.collect(Collectors.toList()).toFuture()

这段代码会把每个Int元素转换成CompletableFuture<Int>，最终得到CompletableFuture<List<CompletableFuture<Int>>>——这种嵌套的Future结构GraphQL无法解析，导致返回null。而第二个版本直接将Flux收集为List<Int>的Future，符合GraphQL对列表类型的预期。

验证步骤 #

1. 更新GraphQL Schema，将asynchronouslyDoFluxInt改为[Int]类型
2. 使用修正后的CustomFunctionDataFetcher
3. 重新执行查询，此时Flux收集的List<Int>会被正确序列化为GraphQL列表类型，报错消失

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