Kotlin流与Java互操作回调

Question

我一直在寻找一个适合的解决方案或者最佳实践，以便在我想要使用Kotlin Flows和普通回调函数时使用。我的用例是编写一个使用Kotlin Flow内部的Kotlin库，并且我必须假设用户会使用Java等语言。因此，我认为最好的解决方案是重载基本的回调接口到我的Flow方法中，在 collect 中进行调用，类似于以下代码：

class KotlinClass {

    interface Callback {
        fun onResult(result: Int)
    }

    private fun foo() = flow {
        for (i in 1..3) {
            emit(i)
        }
    }

    fun bar(callback: Callback) {
        runBlocking {
            foo().collect { callback.onResult(it) }
        }
    }

  private fun main() {
    bar(object : Callback {
        override fun onResult(result: Int) {
            TODO("Not yet implemented")
        }
    })
}

在我的Java应用程序中，我可以简单地像这样使用它：

public class JavaClass {

    public void main() {
        KotlinClass libraryClass = new KotlinClass();
        libraryClass.bar(new KotlinClass.Callback() {
            @Override
            public void onResult(int result) {
                // TODO("Not yet implemented")
            }
        });
    } 
}

我不确定应该怎么做，因为我希望我的Kotlin库可以在Java和Kotlin中使用并且使用Flows的方式良好。

我看到了callbackFlow，但是那似乎只适用于将基于回调的API转换成流？因为我对Kotlin和Flows很陌生，请原谅如果我的问题由于缺少一些基本概念而有缺陷。

- edhair

Answer 1

我会让Java客户端更加掌握流程。我会在您的回调接口中添加一个onStart和onCompletion方法。此外，我会使用自己的CoroutineScope-也许可以从Java客户端进行自定义。而且，我不会在Kotlin函数中阻塞调用线程-不会使用runBlocking。

@InternalCoroutinesApi
class KotlinClass {
    val coroutineScope = CoroutineScope(Dispatchers.Default)

    interface FlowCallback {
        @JvmDefault
        fun onStart() = Unit

        @JvmDefault
        fun onCompletion(thr: Throwable?) = Unit
        fun onResult(result: Int)
    }

    private fun foo() = flow {
        for (i in 1..3) {
            emit(i)
        }
    }

    fun bar(flowCallback: FlowCallback) {
        coroutineScope.launch {
            foo().onStart { flowCallback.onStart() }
                .onCompletion { flowCallback.onCompletion(it) }
                .collect { flowCallback.onResult(it) }
        }
    }

    fun close() {
        coroutineScope.cancel()
    }    
}

现在Java客户端完全控制如何启动、收集和取消流程。例如，您可以使用闩锁来等待完成，设置超时并取消协程范围。这看起来一开始可能需要写很多代码，但通常您会需要这种灵活性。

public class JavaClass {
    public static void main(String[] args) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(1);
        KotlinClass libraryClass = new KotlinClass();
        libraryClass.bar(new KotlinClass.FlowCallback() {
            @Override
            public void onCompletion(@Nullable Throwable thr) {
                latch.countDown();
            }


            @Override
            public void onResult(int result) {
                System.out.println(result);
            }
        });

        try {
            latch.await(5, TimeUnit.SECONDS);
        } finally {
            libraryClass.close();
        }
    }
}

- Rene

---

感谢您的回答，但我还不完全理解countdownLatch部分，我是否真的需要从我的Java库中关闭coroutineScope，或者当它在Kotlin库中完成时，我可以直接终止它？ - edhair

你也可以从 Kotlin 中关闭 coroutineScope，对于阻止调用线程也是如此。但是你的问题是，什么样的 API 才是好的。如果你在自己的库中这样做，就会对客户端使用产生很大的影响。从我的角度来看，更好的方式是让客户端自己决定阻塞线程是否是个好主意以及超时时间应该多长。 - Rene

问题是我真的想要隐藏Kotlin库中的所有线程，并且Java库应该只在每个收集到的Flow结果的主线程上获得回调。 - edhair

感谢您提供有关 Kotlin Flow 的 Java 互操作性的这些想法 :heart: - mochadwi

Answer 2

您不需要在Kotlin代码中创建接口。您可以像这样定义bar：

 fun bar(callback: (Int) -> Unit) {
     runBlocking {
         foo().collect { callback(it) }
     }
 }

从Java代码中，您可以这样调用该函数：

public class JavaClass {

    public static void main(String[] args) {
        KotlinClass libraryClass = new KotlinClass();
        libraryClass.bar(v -> { System.out.println(v); return Unit.INSTANCE; });
    } 
}

- Diego Marin Santos

---

这是一个有效的建议，但我认为在我的Java代码中使用return Unit.INSTANCE会像这里描述的那样是一种不好的做法：https://developer.android.com/kotlin/interop？要在我的Java代码中使用`Unit`类，我需要添加一些Kotlin依赖项或其他东西吗？因为我无法直接使用它。 - edhair

目前无法定义一个参数类型，以便从Java和Kotlin中使用lambda，使其在两种语言中都感觉自然。当前的建议是优先选择函数类型，尽管当返回类型为Unit时，Java的体验会降级。如果您在Java项目中使用kotlin，则应该已经将Kotlin stdlib作为依赖项包含在内，不是吗？如果您的库将仅从Java中使用，则还可以尝试最后一个建议“在Java中定义命名的SAM接口”。 - Diego Marin Santos

Answer 3

如果有人想要一个通用解决方案，这里是我们从@rene的答案中进行改进的版本。

1. 接受任何类型
2. 可配置的coroutineScope

// JavaFlow.kt
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.InternalCoroutinesApi
import kotlinx.coroutines.cancel
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.onCompletion
import kotlinx.coroutines.flow.onStart
import kotlinx.coroutines.launch

@InternalCoroutinesApi
class JavaFlow<T>(
    private val coroutineScope: CoroutineScope = CoroutineScope(Dispatchers.Default)
) {

    interface OperatorCallback <T> {
        @JvmDefault
        fun onStart() = Unit

        @JvmDefault
        fun onCompletion(thr: Throwable?) = Unit
        fun onResult(result: T)
    }

    fun collect(
        flow: Flow<T>,
        operatorCallback: OperatorCallback<T>,
    ) {
        coroutineScope.launch {
            flow
                .onStart { operatorCallback.onStart() }
                .onCompletion { operatorCallback.onCompletion(it) }
                .collect { operatorCallback.onResult(it) }
        }
    }

    fun close() {
        coroutineScope.cancel()
    }
}

Java调用方：

// code omitted...
new JavaFlow<File>().collect(
        // compressImageAsFlow is our actual kotlin flow extension
        FileUtils.compressImageAsFlow(file, activity),
        new JavaFlow.OperatorCallback<File>() {
            @Override
            public void onResult(File result) {
                // do something with the result here
                SafeSingleton.setFile(result);
            }
        }
);


// or using lambda with method references
// new JavaFlow<File>().collect(
//        FileUtils.compressImageAsFlow(file, activity),
//        SafeSingleton::setFile
// );

// Change coroutineScope to Main
// new JavaFlow<File>(CoroutineScopeKt.MainScope()).collect(
//        FileUtils.compressImageAsFlow(file, activity),
//        SafeSingleton::setFile
// );

OperatorCallback.onStart 和 OperatorCallback.onCompletion 是可选的，根据需要进行重写。

- mochadwi