Котлин Coroutines против CompletableFuture

Question

Может кто-нибудь объяснить мне, почему люди должны использовать сопрограммы? Есть ли пример кода сопрограммы, который показывает лучшее время завершения для обычного параллельного кода Java (без магической функции delay (), никто не использует delay() в производстве) ?

В моем личном примере сопрограммы(строка 1) не подходит для кода Java (строка 2). Может быть, я сделал что-то не так?

Пример:

import kotlinx.coroutines.*
import java.time.Instant
import java.util.concurrent.CompletableFuture
import java.util.concurrent.Future

@ExperimentalCoroutinesApi
fun main() = runBlocking {
    val begin = Instant.now().toEpochMilli()
    val jobs = List(150_000) {
        GlobalScope.launch { print(getText().await()) } // :1
//        CompletableFuture.supplyAsync { "." }.thenAccept { print(it) } // :2
    }
    jobs.forEach { it.join() }
    println(Instant.now().toEpochMilli() - begin)
}

fun getText(): Future<String> {
    return CompletableFuture.supplyAsync {
        "."
    }
}

@ExperimentalCoroutinesApi
suspend fun <T> Future<T>.await(): T = suspendCancellableCoroutine { cont ->
    cont.resume(this.get()) {
        this.cancel(true)
    }
}

Дополнительный вопрос:

Почему я должен создать эту оболочку сопрограммы await()? Кажется, не улучшает сопрограммную версию кода, в противном случае get() метод жалуется на inappropriate blocking method call?

Answer 1

Целью сопрограмм является не «лучшее время завершения». Честно говоря, цель, к которой это добивается успеха, честно, состоит в том, чтобы сопрограммы были проще в использовании .

Тем не менее, то, что вы сделали в своем коде, вовсе не - хороший способ сравнить скорость двух альтернативных подходов. Сравнение скорости вещей в Java и получение реалистичных результатов чрезвычайно сложно , и вы должны прочитать Как мне написать правильный микро-бенчмарк в Java? как минимум, прежде чем пытаться это сделать. То, как вы в настоящее время пытаетесь сравнить два фрагмента кода Java, будет обманывать вас относительно реалистичного поведения вашего кода.

Чтобы ответить на ваш дополнительный вопрос, ответ таков: не должен создавать этот await метод. Вы не должны использовать get() - или java.util.concurrent.Future - с кодом сопрограммы, будь то suspendCancellableCoroutine или другим способом. Если вы хотите использовать CompletableFuture, используйте предоставленную библиотеку для взаимодействия с ней из кода сопрограммы.

Answer 2

Вот очищенная версия вашего кода, которую я использовал для бенчмаркинга. Обратите внимание, что я удалил print из измеренного кода, потому что сама печать является тяжеловесной операцией, включающей мьютексы, JNI, блокировку выходных потоков и т. Д. Вместо этого я обновляю переменную volatile.

import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.future.await
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import java.lang.Thread.sleep
import java.util.concurrent.CompletableFuture
import java.util.concurrent.CompletionStage
import java.util.concurrent.TimeUnit.NANOSECONDS

@Volatile
var total = 0

@ExperimentalCoroutinesApi
fun main() = runBlocking {
    println("Warmup")
    measure(20_000)
    println("Measure")
    val begin = System.nanoTime()
    measure(40_000)
    println("Completed in ${NANOSECONDS.toMillis(System.nanoTime() - begin)} ms")
}

fun getText(): CompletableFuture<Int> {
    return CompletableFuture.supplyAsync {
        sleep(1)
        1
    }
}

suspend fun measure(count: Int) {
    val jobs = List(count) {
        GlobalScope.launch { total += getText().await() } // :1
//        getText().thenAccept { total += it } // :2
    }
    jobs.forEach { it.join() }
}

Мой результат составляет 6,5 секунды. для случая номер один против 7 секунд для случая номер два. Это разница в 7%, и она, вероятно, очень специфична для этого конкретного сценария, а не то, что вы обычно будете видеть как разницу между этими двумя подходами.

Причина выбора сопрограмм вместо CompletionStage программирования определенноне о тех 7%, а о огромной разнице в удобстве. Чтобы понять, что я имею в виду, я предлагаю вам переписать функцию main, вызвав просто computeAsync, без использования future.await():

suspend fun main() {
    try {
        if (compute(1) == 2) {
            println(compute(4))
        } else {
            println(compute(7))
        }
    } catch (e: RuntimeException) {
        println("Got an error")
        println(compute(8))
    }
}

fun main_no_coroutines() {
    // Let's see how it might look!
}

fun computeAsync(input: Int): CompletableFuture<Int> {
    return CompletableFuture.supplyAsync {
        sleep(1)
        if (input == 7) {
            throw RuntimeException("Input was 7")
        }
        input % 3
    }
}

suspend fun compute(input: Int) = computeAsync(input).await()

Answer 3

Мои 2 версии метода compute без переписывания методов подписи. Я думаю, что я понял вашу точку зрения. С сопрограммами мы пишем сложный параллельный код в привычном последовательном стиле. Но оболочка сопрограммы await не выполняет эту работу из-за техники приостановки, она просто реализует ту же логику, что и я.

import java.lang.Thread.sleep
import java.util.concurrent.CompletableFuture

fun main() {
    try {
        if (compute(1) == 2) {
            println(compute(4))
        } else {
            println(compute(7))
        }
    } catch (e: RuntimeException) {
        println("Got an error")
        println(compute(8))
    }
}

fun compute(input: Int): Int {
    var exception: Throwable? = null
    val supplyAsync = CompletableFuture.supplyAsync {
        sleep(1)
        if (input == 7) {
            throw RuntimeException("Input was 7")
        }
        input % 3
    }.exceptionally {
        exception = it
        throw it
    }
    while (supplyAsync.isDone.not()) {}
    return if (supplyAsync.isCompletedExceptionally) {
        throw exception!!
    } else supplyAsync.get()
}

fun compute2(input: Int): Int {
    try {
        return CompletableFuture.supplyAsync {
            sleep(1)
            if (input == 7) {
                throw RuntimeException("Input was 7")
            }
            input % 3
        }.get()
    } catch (ex: Exception) {
        throw ex.cause!!
    }
}

Answer 4

После переключения на эту библиотеку kotlinx-coroutines-jdk8 и добавления sleep (1) в мою функцию getText()

import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.future.await
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import java.lang.Thread.sleep
import java.time.Instant
import java.util.concurrent.CompletableFuture

fun main() = runBlocking {
    val begin = Instant.now().toEpochMilli()
    val jobs = List(150_000) {
        GlobalScope.launch { print(getText().await()) } // :1
//        getText().thenAccept { print(it) } // :2
    }
    jobs.forEach { it.join() }
    println(Instant.now().toEpochMilli() - begin)
}

fun getText(): CompletableFuture<String> {
    return CompletableFuture.supplyAsync {
        sleep(1)
        "."
    }
}

я сделал версию сопрограммы быстрее, чем версию java !! ! По-видимому, этот дополнительный слой сопрограммы становится оправданным, когда есть некоторая задержка.