今天开始要通过阅读深入理解Kotlin协程这本书来学习Kotlin协程的原理以及用途,并且会把阅读笔记在本站分享出来,一般我会边读边写,而不是读完之后一口气写出来,所以有时候可能会出现一些前后逻辑不通顺的情况,我在写完之后会通读文章进行勘误,然而由于我才学疏浅,遗漏是不可避免的事情,如果你发现文中有错误的部分,欢迎在评论区指出

什么是异步

异步是一个比较模糊的抽象概念,他与同步的概念相反,指令不按照定义时的顺序执行称之为异步,反之为同步,其中异步可以伴随着并行或者并发1,但是异步并不等同于并行,例如在单线程程序中也可以实现异步,但是无法实现并行。异步的实现可以有很多种,比如一段典型的多线程代码:

    thread {
        Thread.sleep(2000)
        println("B")
    }
    println("A")

其中println("A")在定义的顺序上后于println("B"),可实际执行时A却先于B被打印在控制台上,这就是典型的异步代码

异步回调

通常我们在使用多线程代码时经常会遇到一些比较尴尬的情况,比如我们启动了一个线程A执行耗时的I/O操作,可是接下来的逻辑中我们又需要这个I/O操作的结果,一个常见的情景是这样的

    val threadPool = Executors.newSingleThreadExecutor()
    // start a new thread to do some long running tasks so current thread can keep doing other works such as render UI without blocking
    val future = threadPool.submit<Int> {
        Thread.sleep(2000)
        1024
    }
    // assume we need the internal value of future here, e.g. update the value of a inputbox...

在这种情况下,一般来说我们可以调用future.get()来获取到结果,可是问题在于Future::get()方法会阻塞当前的线程直到其计算完成并成功获取到结果(其实想想很好理解,不阻塞当前线程你怎么能得到结果呢?好比你在银行柜台前办事,进行一个阶段之前必须等待上一阶段完成),这违背了多线程的初衷,如果这是一个GUI程序的话,那么UI线程就会在get()方法处阻塞,从而造成界面假死,极大的降低用户体验。

想象一下,在一个UI应用程序中常见的情景是UI线程向后台线程提交一个任务,之后等到任务完成之后通知UI去根据任务的结果更新某些区域(比如修改文本框的值),如果放在现实生活中,还用上面银行的例子的话,就是这样的:
1. 我向银行工作人员提交某些申请,并且留下我的联系方式,告诉他们处理完通知我
2. 银行工作人员开始处理我的申请,同时我去干别的事情,比如上班上学
3. 等到银行工作人员处理完我的申请之后,通过我的联系方式来告诉我"我们已经处理完您的申请,请来我行办理下一步的业务"
可以看到,这是我们日常生活中最常用的方式,抽象出来之后可以划分为几步:假设"我"是主线程,"银行工作人员"是后台线程,而工作人员处理我提交的申请是一个耗时任务,在第一步中,主线程通过留下联系方式来注册一个回调(Callback,这个英文名称非常的形象),这个回调告诉了后台线程任务执行完之后该怎么做;接下来在第二步中,后台线程执行耗时操作,同时主线程继续做自己该做的事情;第三步,耗时任务完成后,后台线程通过回调(也就是我的联系方式)通知主线程,告诉主线程"耗时任务已经执行完毕",这种方式就叫做异步回调,我们可以用如下的代码实现它:(如果你看不懂这部分的代码,说明你对于kotlin基础语法掌握不熟,请前往Kotlin官方文档仔细阅读)

    // 假设scheduleTask是银行管理业务的工作人员
    // task参数相当于我们告诉工作人员我们要办理什么业务
    // callback参数即为我们注册的回调,告诉工作人员业务办理完怎么通知我
    fun scheduleTask(task: () -> Int, callback: (Int) -> Unit) {
        println("Bank: The staff has accepted your request")
        val threadPool = Executors.newSingleThreadExecutor()
        threadPool.submit {
            val result = task() // 银行(后台线程)执行业务
            callback(result) // 银行(后台线程)通过注册的回调通知我(主线程)
        }
    }
    ....
    // 我的联系方式(回调)
    val callback: (Int) -> Unit = {
        println("Bank: Your request is finished with card ID: $it")
    }
    // 要办理的业务(耗时操作)
    val task: () -> Int = {
        Thread.sleep(5000)
        200
    }
    println("Me: I'm going to the bank to ask for a card")
    // 告诉银行去执行我的业务(让后台线程去执行耗时操作),同时告诉他们我的联系方式(注册一个回调)
    scheduleTask(task, callback)
    println("Me: I have submitted my request")
    // 接下来我可以继续做我要做的事情,不用自己关心业务办完没有,因为办完之后银行会通知我
    println("Me: I'm having lunch now")
    println("Me: I'm at school now")
    println("Me: I'm having a party now")

程序输出:

    DC: I'm going to the bank to ask for a card
    Bank: The staff has accepted your request
    DC: I have submitted my request
    DC: I'm having lunch now
    DC: I'm at school now
    DC: I'm having a party now
    Bank: Your request is finished with card ID: 200

我们使用这么一段代码来实现了生活中常见的"异步与回调",而这种设计模式同样也广泛的应用在各种程序中,虽然这种模式能够较为方便的处理异步的需求,但是其缺点也显而易见,与现实中不同,这里的异步与回调代码脱离了主控制流,如果我们需要在callback中执行同样的耗时——等待获取结果的操作,代码会变成什么样子呢?

    fun asyncRun(task: () -> Unit, callback: () -> Unit) {
        thread {
            task()
            callback()
        }
    }

    asyncRun({
        Thread.sleep(2000)
        println("A")
    }) {
        asyncRun({
            Thread.sleep(3000)
            println("B")
        }) {
            asyncRun({
                Thread.sleep(5000)
                println("C")
            }) {
                println("D")
            }
        }
        println("E")
    }

Java提供了哪些异步方法?

Future<T>接口

Future<T>接口在java中扮演了非常重要的角色,大部分的java并发框架(CompletableFuture<T>, Executor等)中都有它的身影,顾名思义,Future<T>代表了一个会在未来完成的任务,我们可以通过Future::get()方法获取到任务的执行结果,如果任务尚未执行完毕,那么对于get()的调用将会阻塞当前线程直到任务执行完毕

CompletableFuture<T>

同样是回调,相比起之前的asyncRunCompletableFuture<T>提供了更加优雅的办法,写过JavaScript的人应该很清楚Promise与它的then()方法,CompletableFuture<T>在Java里提供了一套类似的API,你可以通过诸如下面的方法实现上面的回调地狱: (如果你看不懂如下代码,说明你不理解什么是Java的Lambda表达式,请自行Google/Baidu,在此不再赘述)

    CompletableFuture.runAsync(() -> {
        Thread.sleep(2000);
        System.out.println("A");
    }).thenAccept(() -> {
        Thread.sleep(3000);
        System.out.println("B");
    }).thenAccept(() -> {
        Thread.sleep(5000);
        System.out.println("C");
    }).thenAccept(() -> {
        System.out.println("D");
    }) /* 我还可以再加上一段带有返回值的 */ .thenApply(() -> {
        System.out.println("E with return value");
        return 200;
    }).thenAccept(System.out::println) // 注意这里的结果获取是在thenAccept里而非外部,已经脱离了主控制流

看上去是不是清爽多了?你没必要把一层回调写在另一层回调里面,而是可以通过链式调用的方法,把链式调用传入的函数作为上一个调用的函数的回调

Future<T>CompletableFuture<T>的缺陷

从上文中可以很容易的发现,不管是Future<T>还是CompletableFuture<T>,他们共同的缺陷就是要么阻塞当前线程获取结果,要么就在另一个控制流中获取结果,比如上文中的CompletableFuture,其结果的获取发生在thenAccept内部,如果你想要在外部直接获取结果,例如int result = CompletableFutre...那么就要调用CompletableFuture::get()方法,把代码写成int result = CompletableFutre...get(),然而这里的get()同样也会阻塞当前线程2,这样就导致虽然代码可以不如回调地狱那般可怕,但是依然不尽如人意,那么,该如何做到不阻塞当前线程的同时,又能让结果在主控制流程获取呢?

async/await Asynchronous Programming Pattern

async/await异步编程模式完美的解决了上述问题,在C#中,你可以用Task<T>(相当于C#版本的CompletableFuture<T>)来复现上述代码:

    Task.Run(() => {
        Thread.Sleep(2000);
        Console.WriteLine("A")
    }).ContinueWith(() => {
        Thread.Sleep(3000);
        Console.WriteLine("B")
    }).ContinueWith(() => {
        Thread.Sleep(5000);
        Console.WriteLine("C");
    }).ContinueWith(() => {
        Console.WriteLine("D")
    }).ContinueWith(() => {
        Console.WriteLine("E with return value");
        return 200;
    }).ContinueWith(Console.WriteLine)

看上去似乎并没有什么区别?只不过是移植了一下CompletableFuture<T>?
实则不然,C#的Task<T>是一个Awaitable的类,也就是其中包含了GetAwaiter()方法的类,这些类可以使用async/await运算符来让异步逻辑同步化,什么意思,就是可以取出回调操作,做到不阻塞线程的同时在主控制流获取结果,比如上面的代码可以改写成下面这样:

    await Task.Delay(2000);
    Console.WriteLine("A");
    await Task.Delay(3000);
    Console.WriteLine("B");
    await Task.Delay(5000);
    Console.WriteLine("C");
    Console.WriteLine("D");
    Console.WriteLine("E with return value");
    Console.WriteLine(200);

是不是代码变得一目了然了?在C#中,await运算符的表现可以理解为隐式的把位于它后面的代码转换为异步回调,也就是:

    Task.Run(() => DoSomeWork()).ContinueWith(result => Accept(result))

等同于

    var result = await DoSomeWork();
    Accept(result);

可以进一步简写成

    Accept(await DoSomeWork())

当然,转换的过程就并非这么简单和一目了然了,C#内部会将await/async代码转换为一个异步状态机,通过状态机的状态流转决定该如何控制代码执行流程,其中还包含了TaskSchedulerSynchronizationContext对线程的调度,具体的解释如果有兴趣的话可以看msdn上的blog,我在这里只是演示了async/await异步编程模式是如何同时做到非阻塞和不脱离主控制流程的,当下的许多支持异步的语言都选择了async/await作为其实现方式,包括JavaScript,C#,Python,Rust以及C++20(严格来说C++20的co_yieldco_await与前文所提到的都不同,C++20提供了更加自由和更加底层的方式去操作一个协程,就像kotlin做到的那样),看到这里,也许你觉得Kotlin也应该用这种最广为人知也最容易上手的方式,但实际上Kotlin虽然提供了asyncawait,却并没有使用上述语言中所使用的关键字实现,Kotlin在不加入额外的关键字的情况下实现了async/await模式,具体原理会在以后读到书的具体位置的时候写Blog记录下来。


  1. 深入理解计算机系统 p17指出,并发指具有多个活动的系统,而并行指使用并发让一个系统运作得更快,更具体的来说,并发指两个操作可以同时存在,而并行指的是两个操作可以同时执行,在单核心CPU上,并发可以是CPU交替执行两个进程的指令,而在多核心CPU上,并行则是并发的一种实现 ↩︎
  2. 事实上CompletableFuture::get()实际上是继承自Future::get(),而Javadoc中对于Future::get()的描述如下:Waits if necessary for the computation to complete, and then retrieves its result.,所以实际上这里的get()操作阻塞是合情合理的 ↩︎

乱我心者,今日之日多烦忧