Kotlin Coroutine

// 定义两个Callable来异步执行方法
Callable<String> getUserName = () -> {
    // 模拟调用耗时方法获取用户名
    Thread.sleep(1000); 
    return "John";
};

Callable<String> getCompany = () -> {
    // 模拟调用耗时方法获取公司名
    Thread.sleep(1000);
    return "Doe Corp."; 
};

// 使用ExecutorService执行两个Callable并获取Future
ExecutorService executor = Executors.newFixedThreadPool(2);
Future<String> nameFuture = executor.submit(getUserName);
Future<String> companyFuture = executor.submit(getCompany);

// 在主线程中获取结果并合并
String name = nameFuture.get();    
String company = companyFuture.get();
String info = name + ", " + company;
System.out.println(info); // John, Doe Corp.

executor.shutdown();

// 定义两个异步操作
CompletableFuture<String> getUserName = CompletableFuture.supplyAsync(() -> {
    Thread.sleep(1000);
    return "John";
});

CompletableFuture<String> getCompany = CompletableFuture.supplyAsync(() -> {
    Thread.sleep(1000);
    return "Doe Corp.";
});

// 使用thenCombine合并两个异步操作的结果
CompletableFuture<String> combined = getUserName.thenCombine(getCompany, (name, company) -> name + ", " + company);

// 获取最终结果
String result = combined.get(); 
System.out.println(result); // John, Doe Corp.

suspend fun getUserName(): String {
    delay(1000)  // 模拟调用耗时方法
    return "John"
}

suspend fun getCompany(): String {
    delay(1000)  // 模拟调用耗时方法
    return "Doe Corp."
}

fun main() {
    // 使用async并发调用两个suspend函数
    val name = async { getUserName() }
    val company = async { getCompany() }
    
    // 使用await等待两个任务完成并获取结果
    val nameResult = name.await()  
    val companyResult = company.await()
    
    println("$nameResult, $companyResult") // John, Doe Corp. 
}

// 或者使用awaitAll
suspend fun fetchTwoDocs() =        // called on any Dispatcher (any thread, possibly Main)
    coroutineScope {
        val deferreds = listOf(     // fetch two docs at the same time
            async { fetchDoc(1) },  // async returns a result for the first doc
            async { fetchDoc(2) }   // async returns a result for the second doc
        )
        deferreds.awaitAll()        // use awaitAll to wait for both network requests
        // The awaitAll function should be preferred over map { it.await() }
    }

val time = measureTimeMillis {
    val one = async { doSomethingUsefulOne() }
    val two = async { doSomethingUsefulTwo() }
    println("The answer is ${one.await() + two.await()}")
}
println("Completed in $time ms")

val time = measureTimeMillis {
    val one = async(start = CoroutineStart.LAZY) { doSomethingUsefulOne() }
    val two = async(start = CoroutineStart.LAZY) { doSomethingUsefulTwo() }

    one.start() // 不会阻塞，直接下一行执行
    two.start() 
    println("The answer is ${one.await() + two.await()}") //注意, 如果没有上面两个start的话, 那么这两个await是先后调用，导致两个协程顺序执行而不是异步
}
println("Completed in $time ms")

fun main() = runBlocking<Unit> {
    try {
        failedConcurrentSum()
    } catch(e: ArithmeticException) {
        println("Computation failed with ArithmeticException")
    }
}

suspend fun failedConcurrentSum(): Int = coroutineScope { //coroutineScope函数创建一个新的scope 
    val one = async<Int> { 
        try {
            delay(Long.MAX_VALUE) // Emulates very long computation
            42
        } finally {
            println("First child was cancelled")
        }
    }
    val two = async<Int> { 
        println("Second child throws an exception")
        throw ArithmeticException()
    }
    one.await() + two.await()
}

suspend fun doSomething() {
    withContext(Dispatchers.IO) {
        // 在IO上下文中执行IO密集型代码
        doNetworkRequest() 
    }
}

suspend fun doSomething() {
    val result = withContext(Dispatchers.Default) {
        // 在默认上下文中进行计算
        calculateResult() 
    }
    withContext(Dispatchers.Main) {
        // 在主线程中更新UI
        updateUI(result)
    }
}

suspend fun doSomething() {
    val job1 = GlobalScope.launch(Dispatchers.IO) {
        // ...
    } 
    val job2 = GlobalScope.launch(Dispatchers.Main) {
        // ...
    }
    withContext(Dispatchers.IO) {
        job1.join()   // 等待IO上下文的任务结束
    }
    withContext(Dispatchers.Main) {
        job2.join()   // 等待主线程的任务结束 
    } 
}

suspend fun doSomething() {
    withContext(NonCancellable) { // 使用NonCancellable上下文
        // 这里的代码块不会被取消
        doSomething()
    }
    // ...
}

// async
fun asyncDemo() = runBlocking {
    println("I am working")
    val opOne = async(IO) { operationOne() }.await() //注意 这里会阻塞等到operationOne返回才能继续下一行执行
    val opTwo = async(IO) { operationTwo() }.await()
    println("Done working.")
    println("The multiplied result is ${opOne * opTwo}")
}

// withContext
fun withContextDemo() = runBlocking {
    println("I am working")
    val opOne = withContext(IO) { operationOne() }
    val opTwo = withContext(IO) { operationTwo() }
    println("Done working.")
    println("The multiplied result is ${opOne * opTwo}")
}

val result = withContext(IO) { operationOne() + operationTwo() }

fun simple(): Flow<Int> = flow { // flow builder, no suspend keyword before fun
    for (i in 1..3) {
        delay(1000) // pretend we are doing something useful here
        emit(i) // emit next value
    }
}

// Collect the flow
simple().collect { value -> println(value) }
// We can replace delay with Thread.sleep in the body of simple's flow { ... } and see that the main thread is blocked in this case.

fun simple(): Flow<Int> = flow { 
    for (i in 1..3) {
        delay(100)          
        println("Emitting $i")
        emit(i)
    }
}

fun main() = runBlocking<Unit> {
    withTimeoutOrNull(250) { // Timeout after 250ms 
        simple().collect { value -> println(value) } 
    }
    println("Done")
}
// only collected 1 2 

fun simple(): Flow<Int> = flow {
    for (i in 1..3) {
        Thread.sleep(100) // pretend we are computing it in CPU-consuming way
        log("Emitting $i")
        emit(i) // emit next value
    }
}.flowOn(Dispatchers.Default) // RIGHT way to change context for CPU-consuming code in flow builder

fun main() = runBlocking<Unit> {
    simple().collect { value ->
        log("Collected $value") 
    } 
}            

val time = measureTimeMillis {
    simple() // 100ms for each element
        .buffer() // buffer emissions, don't wait
        .collect { value -> 
            delay(300) // pretend we are processing it for 300 ms
            println(value) 
        } 
}   
println("Collected in $time ms")

val channel = Channel<Int>()
launch {
    for (x in 1..5) channel.send(x * x)
    channel.close() // we're done sending
}
// here we print received values using `for` loop (until the channel is closed)
for (y in channel) println(y)
println("Done!")

public fun <R> start(start: CoroutineStart, receiver: R, block: suspend R.() -> T) {
    start(block, receiver, this) //这里实际调用的是CoroutineStart.invoke方法。this指的是当前coroutine
}