kotlin - 如何将Java阻塞函数转换为可取消挂起函数？

Question

Kotlin suspend 函数按照惯例应该是非阻塞的 (1)。我们经常有依赖于 java 线程中断机制的旧 Java 代码，我们不能(不想)修改 (2):

public void doSomething(String arg) {
    for (int i = 0; i < 100_000; i++) {
        heavyCrunch(arg, i);
        if (Thread.interrupted()) {
            // We've been interrupted: no more crunching.
            return;
        }
    }
}

调整此代码以在协程中使用的最佳方法是什么？

版本 A: Not Acceptable ，因为它将在调用者线程上运行代码。所以它会违反 “挂起函数不会阻塞调用线程” 约定:

suspend fun doSomething(param: String) = delegate.performBlockingCode(param)

版本 B:更好，因为它会在后台线程中运行阻塞函数，因此它不会阻塞调用者线程(除非调用者偶然使用来自 Dispatchers.Default 的相同线程线程池)。但是协程作业取消不会中断依赖于线程中断的 performBlockingCode()。

suspend fun doSomething(param: String) = withContext(Dispatchers.Default) {
    delegate.performBlockingCode(param)
}

版本 C:目前我认为这是唯一可以让它工作的方法。思路是通过Java机制将阻塞函数转为非阻塞函数，然后使用suspendCancellableCoroutine (3)将异步方法转为suspend函数:

private ExecutorService executor = Executors.newSingleThreadExecutor();

public Future doSomethingAsync(String arg) {
    return executor.submit(() -> {
        doSomething(arg);
    });
}

suspend fun doSomething(param: String) = suspendCancellableCoroutine<Any> { cont ->
    try {
        val future = delegate.doSomethingAsync(param)
    } catch (e: InterruptedException) {
        throw CancellationException()
    }
    cont.invokeOnCancellation { future.cancel(true) }
}

如下所述，上面的代码将无法正常工作，因为没有调用 continuation.resumeWith()

版本 D:使用 CompletableFuture:它提供了一种在可完成完成时注册回调的方法:thenAccept

private ExecutorService executor = Executors.newSingleThreadExecutor();

public CompletableFuture doSomethingAsync(String arg) {
    return CompletableFuture.runAsync(() -> doSomething(arg), executor);
}

suspend fun doSomething(param: String) = suspendCancellableCoroutine<Any> { cont ->
    try {
        val completableFuture = delegate.doSomethingAsync(param)
        completableFuture.thenAccept { cont.resumeWith(Result.success(it)) }
        cont.invokeOnCancellation { completableFuture.cancel(true) }
    } catch (e: InterruptedException) {
        throw CancellationException()
    }
}

你知道更好的方法吗？

1. https://docs.oracle.com/javase/tutorial/essential/concurrency/interrupt.html
2. https://medium.com/@elizarov/blocking-threads-suspending-coroutines-d33e11bf4761
3. https://medium.com/@elizarov/callbacks-and-kotlin-flows-2b53aa2525cf

Answer 1

您可以通过 suspend fun kotlinx.coroutines.runInterruptible 包装阻塞代码

它抑制了编译警告和阻塞代码将在取消时抛出 InterruptedException

val job = launch {
  runInterruptible {
    Thread.sleep(500)
  }
}

job.cancelAndJoin() // Cause will be 'java.lang.InterruptedException'

在 org.jetbrains.kotlinx:kotlinx-coroutines-core-jvm:1.4.2 上测试