如何从Java代码中调用Kotlin的Flow方法？

场景重现 #

Kotlin端定义了一个返回Flow的方法：

fun getAllUsers(): Flow<List<User>> =
        userRepositoryImpl.getAllUsers()

Java端尝试调用时直接传递Flow对象会出现类型不兼容问题，原尝试代码：

private Observable<List<User>> getUsers() {
  subscriber.onNext(userViewModel.getAllUsers()); // 此处直接传入Flow会报错
}

可行解决方案 #

Flow是Kotlin协程专属的响应式流类型，Java本身不直接支持，需要借助协程适配库做类型转换，以下是几种常用方案：

1. 转换为RxJava Observable（推荐，适配RxJava项目） #

如果你的Java项目基于RxJava开发，引入kotlinx-coroutines-rx3（RxJava3版本，RxJava2用kotlinx-coroutines-rx2）依赖后，就能把Flow直接转成Observable：

import kotlinx.coroutines.rx3.FlowKt;
import io.reactivex.rxjava3.core.Observable;

private Observable<List<User>> getUsers() {
    // 调用FlowKt.asObservable完成类型转换
    return FlowKt.asObservable(userViewModel.getAllUsers());
}

转换后的Observable可以直接用RxJava的所有操作符处理，完全适配Java端的RxJava生态。

2. 转换为Java CompletableFuture（适配无RxJava的项目） #

如果项目不用RxJava，想基于Java原生异步API处理，可以引入kotlinx-coroutines-jdk8依赖，将Flow转换为CompletableFuture：

• 收集Flow的全部发射结果：

import kotlinx.coroutines.jdk8.FlowKt;
import java.util.concurrent.CompletableFuture;
import java.util.ArrayList;
import java.util.List;

private CompletableFuture<List<List<User>>> getAllUsers() {
    return FlowKt.toList(userViewModel.getAllUsers(), new ArrayList<>());
}

• 仅获取Flow发射的第一个结果：

import kotlinx.coroutines.jdk8.FlowKt;
import java.util.concurrent.CompletableFuture;
import java.util.List;

private CompletableFuture<List<User>> getFirstUserList() {
    return FlowKt.first(userViewModel.getAllUsers());
}

3. 直接在Java中启动协程收集Flow（不推荐，代码繁琐） #

如果一定要直接操作Flow的流式数据，可以通过协程作用域启动收集，但代码会比较冗长：

import kotlinx.coroutines.GlobalScope;
import kotlinx.coroutines.Job;
import kotlinx.coroutines.flow.FlowCollector;
import kotlin.coroutines.Continuation;
import kotlin.coroutines.EmptyCoroutineContext;

private Job collectUsers() {
    return GlobalScope.INSTANCE.launch(EmptyCoroutineContext.INSTANCE, null, () -> {
        userViewModel.getAllUsers().collect(new FlowCollector<List<User>>() {
            @Override
            public Object emit(List<User> value, Continuation<? super Unit> continuation) {
                // 在这里处理每个发射的User列表
                subscriber.onNext(value);
                return Unit.INSTANCE;
            }
        }, null);
    });
}

注意：所有方案都需要确保项目中已引入基础的kotlinx-coroutines-core依赖，再搭配对应的适配库（rx3/jdk8）。

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