fun requestTokenAsync(cb: (Token) -> Unit) { ... }
fun createPostAsync(token: Token, item: Item, cb: (Post) -> Unit) { ... }
fun processPost(post: Post) { ... }

fun postItem(item: Item) {
    requestTokenAsync { token ->
        createPostAsync(token, item) { post ->
            processPost(post)
        }
    }
}

CompletableFuture and Rx

fun requestTokenAsync(): CompletableFuture<Token> { ... }
fun createPostAsync(token: Token, item: Item): CompletableFuture<Post> { ... }
fun processPost(post: Post) { ... }

fun postItem(item: Item) {
    requestTokenAsync()
            .thenCompose { token -> createPostAsync(token, item) }
            .thenAccept { post -> processPost(post) }
            .exceptionally { e ->
                e.printStackTrace()
                null
            }
}

fun requestToken(): Token { ... }
fun createPost(token: Token, item: Item): Post { ... }
fun processPost(post: Post) { ... }

fun postItem(item: Item) {
    Single.fromCallable { requestToken() }
            .map { token -> createPost(token, item) }
            .subscribe(
                    { post -> processPost(post) }, // onSuccess
                    { e -> e.printStackTrace() } // onError
            )
}

协程-同步函数返回

suspend fun requestToken(): Token { ... }   // 挂起函数
suspend fun createPost(token: Token, item: Item): Post { ... }  // 挂起函数
fun processPost(post: Post) { ... }

fun postItem(item: Item) {
    GlobalScope.launch {
        val token = requestToken()
        val post = createPost(token, item)
        processPost(post)
        // 需要异常处理，直接加上 try/catch 语句即可
    }
}

协程恢复resume

协程的挂起通过suspend挂起函数实现，协程的恢复通过Continuation.resumeWith实现。

suspend fun requestToken(): Token { ... }

实际上在 JVM 中更像下面这样：

Object requestToken(Continuation<Token> cont) { ... }

Continuation的定义如下，类似于一个通用的回调接口：

/**
 * Interface representing a continuation after a suspension point that returns value of type `T`.
 */
public interface Continuation<in T> {
    /**
     * Context of the coroutine that corresponds to this continuation.
     */
    public val context: CoroutineContext

    /**
     * Resumes the execution of the corresponding coroutine passing successful or failed [result] as the
     * return value of the last suspension point.
     */
    public fun resumeWith(result: Result<T>)
}

现在再看之前postItem函数：

suspend fun requestToken(): Token { ... }   // 挂起函数
suspend fun createPost(token: Token, item: Item): Post { ... }  // 挂起函数
fun processPost(post: Post) { ... }

fun postItem(item: Item) {
    GlobalScope.launch {
        val token = requestToken()
        val post = createPost(token, item)
        processPost(post)
    }
}

然而，协程内部实现不是使用普通回调的形式，而是使用状态机来处理不同的挂起点，大致的 CPS(Continuation Passing Style) 代码为：

// 编译后生成的内部类大致如下
final class postItem$1 extends SuspendLambda ... {
    public final Object invokeSuspend(Object result) {
        ...
        switch (this.label) {
            case 0:
                this.label = 1;
                token = requestToken(this)
                break;
            case 1:
                this.label = 2;
                Token token = result;
                post = createPost(token, this.item, this)
                break;
            case 2:
                Post post = result;
                processPost(post)
                break;
        }
    }
}

上面代码中每一个挂起点和初始挂起点对应的 Continuation 都会转化为一种状态，协程恢复只是跳转到下一种状态中。挂起函数将执行过程分为多个 Continuation 片段，并且利用状态机的方式保证各个片段是顺序执行的。

协程之间的关系

父协程手动调用`cancel()`或者异常结束，会立即取消它的所有子协程

父协程必须等待所有子协程完成（处于完成或者取消状态）才能完成

子协程抛出未捕获的异常时，默认情况下会取消其父协程。

Executor.asCoroutineDispatcher()

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/java.util.concurrent.-executor/as-coroutine-dispatcher.html

kotlin协程

优秀文章

谷歌开发者

官网doc

和Android Jetpack结合

异步编程模型

异步回调