硬核系列 | 深入剖析 Java 协程

前言

对于大部分Java开发人员而言，协程（Coroutine）是一个相对陌生的概念，但实际上，协程并非是一门崭新的技术，相反，早在1963年协程的概念就已经被正式提出，它的诞生甚至早于线程，后者诞生于1967年。在云原生时代背景下，各个编程语言之间百花斗艳着实热闹，GO语言的成功，让我们重新审视并真正见识到了协程的威力。较为遗憾的是，随着Java15 GA版本的正式发布，Java的设计者们也没能为开发人员在API层面提供对协程的支持。因此，如果我们想在程序中使用协程来提升特定场景下应用程序的执行性能则只能依赖于第三方开源构件（比如：quasar、coroutines、kilim等），或是采用混合编程（比如：Kotlin）的方式进行。在此大家需要注意，由于Java API目前并不提供对协程的支持，因此，无论是使用开源构件还是基于混合编程的方式，语法层面使用协程编程注定会举步维艰。当然，这并不能够成为阻碍我们去了解协程的理由，有朝一日，当Java的原生协程库正式来临之际，必然会重塑大家对并发编程的认识。



本章，我会深入为大家讲解协程的相关概念、在程序中如何应用协程编程，以及为大家分享如何基于字节码增强技术来实现一个简单的协程框架。

Coroutine的优势

在正式为大家深入讲解协程之前，我觉得有必要首先为大家梳理下并发编程的一些前提知识，有助于大家能够更好的理解协程。我们都知道，线程是一种非常昂贵的系统资源，如果线程的执行时间小于其创建和销毁的时间，且在程序中频繁的创建会大幅度增加系统的开销（比如：占用大量物理内存、频繁GC），严重影响程序的执行效率。因此，通常的解决方案是利用线程池这样的池化资源技术来实现资源复用，将线程的生命周期交由线程池来负责管理，避免频繁的创建瞬时线程。大家仔细思考下，线程池中，线程数究竟应该如何设置才算合理？这需要一个科学且合理的量化标准，而不是根据经验来拍脑袋决定，虽然通常情况下大部分开发人员都会选择这么做，但这是不合理的。线程数的设置是需要经过验证的，如果设置得过小，那么CPU的利用率自然不高，反之又会导致CPU不停的切换上下文，从而影响程序的执行性能。在此我为大家提供一个量化线程数的固定计算公式，如下所示：



我们可以统计出线程在执行过程中的CPU时间和I/O操作等时间，将其相加即可得出程序的执行时间。假设CPU时间为500ms，I/O时间同样也为500ms，那么单核CPU的占用率就是50%，理论上2根线程就可以把单核CPU跑满；如果CPU的核心数为16，那么仅需32根线程就可以把所有的CPU跑满，因此量化后的线程数就是32，如果将线程数设置为>32则意义不大。



图1 CPU上下文切换

刚才提及过，程序中如果创建了过多的线程，会导致CPU不停的切换上下文，从而影响程序的执行性能，这是为何呢？在回答这个问题之前，我们首先需要弄清楚究竟什么是CPU上下文？以及为什么需要切换上下文？在Java中，通过java.lang.Thread即可轻松实现并发编程，但实际上，这只是操作系统带给我们的一种感官错觉，多任务并非真的是在同时执行，而是CPU在不停的切换这些任务，轮流地将CPU时间分配给它们，使之看起来就像是真的在同时执行一般。大家思考下，某个任务在执行一段时间后，突然被系统内核切换到别的任务上，之后再被重新调度回来，那么操作系统如何保证前一个任务能够从之前“退出”的指令位置上继续执行？要保证并发任务执行时的正确性和连续性，就必然需要记录任务的状态，即各种CPU寄存器（比如：通用寄存器、程序计数器、状态寄存器等）中的内容（包括但不限于：当前指令位置、下一条指令位置、指令执行结果的各种状态信息，以及中间运算数据等）。任务间的切换，就需要依赖于这些上下文信息；简而言之，就是将上一个任务的CPU上下文信息保存到系统内核中，然后加载新任务的上下文信息至CPU寄存器，待上一个任务被系统内核调度时再重新从系统内核中加载进CPU寄存器内继续执行，同一时刻CPU寄存器仅会独享一个任务，如图1所示。



CPU的上下文切换，通常可以划分为如下3种形式：

• 进程上下文切换；

• 线程上下文切换；

• 中断上下文切换。



本文仅针对进程/线程的上下文切换进行讲解。进程大家应该不会感觉到陌生，它是资源拥有的基本单位，为线程提供了运行环境；而线程包含在进程体内，是内核调度的基本单位，属于进程的执行单元。在Linux操作系统中，根据特权等级，进程的运行空间被划分为：内核空间（Ring 0）和用户空间（Ring 3），如图2所示。Ring 0空间具备最高权限，进程运行在此，即被称为进程的内核态，允许访问所有的物理硬件资源；而Ring 3空间则只能够被允许访问受限资源，进程运行在此，即被称为进程的用户态。在此大家需要注意，如果运行在Ring 3空间中的进程需要访问Ring 0空间中的特权资源，则务必需要通过系统调用陷入到Ring 0空间中。



图2 进程运行空间

进程和线程的上下文切换操作类似，但大部分情况下，线程在发生上下文切换时所带来的性能开销要远低于进程。这是因为，当进程间发生上下文切换时，不但要保存CPU寄存器中的内容，同时还要记录进程资源；但如果线程的切换是发生在同一进程内，由于进程资源被共享，因此仅需记录CPU寄存器中的内容和线程的私有数据（比如：PC寄存器、堆栈等）即可。



当大家清楚究竟什么是CPU上下文，以及为什么执行并发任务需要切换上下文后，我们再回到先前提出的问题上。为何系统内核频繁的切换CPU上下文会影响程序的执行性能？由于上下文切换操作属于CPU密集型任务，通常情况下，需耗费纳秒级别的CPU时间，但如果单位时间内切换频率过高，会导致CPU把大量的时间都花费在上下文信息的保存/恢复上，从而大幅降低任务本身的CPU时间，最终影响程序的吞吐量。



说了这么多关于操作系统原理的相关知识，大家是否感觉到有些枯燥和乏味？这些内容和本文的主题协程究竟有何关系？在I/O密集型场景下，线程数越多，协程的优势就越明显，这也是当下协程被炒冷饭又火起来的主要原因。我之前看过、听过蛮多关于讲解协程原理和使用的文章，很多同学甚至将协程理解为是一种新的并发编程技术，可用于替代线程；这是一种非常错误的观念，虽然我不知道是谁给他们灌输的，但是我认为，哪怕没有协程，在实际的开发过程中，开发人员仍然能够写出优质的代码。协程并不是并发编程领域的未来，它仅仅只是一个补充和完善。



图3 协程的执行模型

协程，又称之为纤程，它是一种语法层面的概念，本身并不属于内核层面。一个协程代表一个具体的任务，一个线程内部可包含一组协程队列，换句话说，协程运行在线程之上，线程是协程的运行环境。刚才提及过，协程非常适用于处理I/O密集型任务，这是因为协程的上下文切换无需由内核调度介入，同时也不会发生系统调用，因此，任务可获得大量的CPU时间。在网络编程（BIO）场景下，服务端往往需要为每一个Sokect都创建一个线程来避免产生I/O阻塞，尽管常见的解决方案是使用更为复杂的NIO模型，但如果基于协程，我们完全可以将由上千个线程完成的任务，替换为1个线程和上千个协程来处理。比如，当协程在执行过程中遇到I/O阻塞时可暂时退出，由用户态的调度器来负责切换到其它就绪协程上，直至I/O阻塞完成后，调度器再重新切换到前一个协程上继续执行。协程的执行模型，如图3所示。在此大家需要注意，和线程的抢占式调度获取CPU时间不同，协程的调度方式见名知意，是基于协作式的，需要由前一个协程主动让出CPU后，其它协程才能够顺利运行。

使用Coroutine

尽管目前Java API并没有为开发人员提供对协程的原生支持，但基于混合编程，或者依赖于第三方开源构件仍然可以在一定程度上满足我们的在业务代码中使用协程的诉求。Kotlin是Java混合编程的首选，值得庆幸的是，Kotlin对协程的支持相对还是比较完善的，底层对各种I/O函数也进行了深度封装；因此，如果想要在Java中使用协程，调用Kotlin的协程函数不失为一计良策。当然，本文的重点是为大家演示如何通过Java原生的方式来实现协程，因此，关于Kotlin的协程演示大家可以自行参考其它的文献资料，本文不再过多进行阐述。接下来，笔者就为大家演示开源协程库coroutines的一些基本使用方式。



在使用coroutines之前，我们首先需要下载其相关构件，示例1-2：

<dependency>
    <groupId>com.offbynull.coroutines</groupId>
    <artifactId>user</artifactId>
    <version>1.5.3</version>
</dependency>



当成功下载好运行coroutines所需的相关构件，以及指定好agent后，接下来要做的事情就是声明协程。coroutines为大家提供了函数式接口com.offbynull.coroutines.user.Coroutine，我们需要实现并重写其run()方法，示例1-3：

Coroutine coroutine = (x) -> {
    System.out.println("A");
    x.suspend();//挂起函数，挂起当前协程
    System.out.println("B");
};



如果想让声明好的协程运行起来，则还需要引入协程调度器CoroutineRunner，并将Coroutine实例传递给它，由调度器来负责协程的上下文切换，示例1-4：

CoroutineRunner r = new CoroutineRunner(coroutine);
r.execute();//启动协程
System.out.println("C");
r.execute();//恢复协程



执行调度器的execute()方法后，会回调Coroutine的run()方法启动协程。当协程在执行过程中触发suspend()方法后，会以不阻塞当前线程为前提主动挂起协程，让出CPU执行权限，并将函数调用堆栈（包括但不限于：局部变量、全局变量，以及程序计数器（字节码行号指示器）等）保存到一个对象中后立即返回。如果后续我们想恢复协程，只需重新调用调度器的execute()方法即可，调度器会负责reload函数调用堆栈，让上一个协程从之前退出的指令位置上继续执行。

抽丝剥茧

在上一小节中，我们学习了协程库coroutines的一些基本使用，那么接下来我再为大家深入分析coroutines的具体实现原理。我们都知道，任意方法在被调用时，JVM都会创建一个独立的栈帧（Frame），以便于维系运行期的各种数据；栈帧的存储空间由当前线程分配在Java虚拟机栈中，每个栈帧内部都包含有局部变量表、操作数栈，以及运行时常量池等信息，如图4所示：



图4 栈帧组成结构

大家思考下，coroutines的调度器是如何做到可以随意暂停Java的方法执行，并通过保存其状态来恢复执行呢？要知道，栈帧的生命周期是随着方法的调用而创建，随着方法的执行结束而销毁，且无论方法是正常结束还是异常退出都算作方法结束；那么调度器在触发suspend()函数那一刻就意味着栈帧已被销毁，再次触发方法调用时一切又将重新开始，也就是说，这一切原本都是无状态的，那么coroutines究竟是如何在语法层面实现协程效果的呢？或许大致你已经猜到了，其实就是通过字节码增强技术实现的，但coroutines究竟对目标类增强了哪些类容呢？源码之前无秘密，我们直接从常量池中获取出增强后的目标类进行反编译，示例1-5：

public void run(Continuation continuation) {
    block8: {
        void c;
        switch (continuation.getMode()) {
            case 0: {
                break;
            }
            case 1: {
                throw new RuntimeException("Unexpected state (saving not allowed at this point)");
            }
            case 2: {
              	// 3、重新触发execute()方法后，恢复方法调用堆栈
                MethodState methodState = continuation.loadNextMethodState();
                Object[] arrobject = methodState.getData();
                switch (methodState.getContinuationPoint()) {
                    case 0: {
                        Object[] arrobject2 = (Object[])arrobject[9];
                        Object[] arrobject3 = (Object[])arrobject[4];
                        Continuation cfr_ignored_0 = (Continuation)arrobject2[0];
                        MyCoroutine this_ = (MyCoroutine)arrobject3[0];
                        continuation = (Continuation)arrobject3[1];
                        continuation.setMode(0);
                        continuation.unloadCurrentMethodState();
                        break block8;
                    }
                    default: {
                        throw new RuntimeException("Unrecognized continuation id");
                    }
                }
            }
            default: {
                throw new RuntimeException("Unrecognized state");
            }
        }
      	// 1、协程执行任务
        System.out.println("A");
      	// 2、触发suspend()中断函数后，保存函数调用堆栈信息后立即返回
        Object[] arrobject = new Object[]{c};
        Object[] arrobject4 = new Object[]{this_, c};
        Object[] arrobject5 = new Object[10];
        arrobject5[4] = arrobject4;
        arrobject5[9] = arrobject;
        c.pushNewMethodState(new MethodState("com.MyCoroutine", -891195303, 0, arrobject5, null));
        c.setMode(1);
        return;
    }
    System.out.println("B");// 4、恢复协程后，从此处开始继续执行
}



示例1-3为增强前的逻辑代码，有效代码总共也就3行，而对其进行增强后，整个方法体内的执行逻辑屹然变得“面目全非”。仔细阅读后不难发现，协程首先会进入到注释1处，执行指令System.out.println("A")。原本下一条指令应该是调用suspend()函数，却被替换为了注释2处的内容，即保存函数调用堆栈后立即返回；尽管栈帧被销毁了，但函数调用堆栈却被完整的保存在了一个上下文对象中。当调度器重新切换到上一个协程后，会进入到注释3处，重新加载协程的上下文信息。一切就绪后，协程最终会进入到注释4处，从之前“退出”的指令位置上继续执行。

实现一个Coroutine框架

目前市面上有关Java开源协程库的实现方式基本大同小异，大家是否想过亲自动手编写一个coroutine框架？尽管重复造轮子意义不大，但在某些情况下，相对于直接阅读源码，造轮子却更能让你加深印象。当然，我并不打算为大家演示如何实现一个比其它开源协程库更加完善的coroutine框架，仅仅只是为了实现协程效果，对此项目感兴趣的同学，可自行重构。想要在语法层面支持协程，除了需要引入字节码增强工具外，还需要依赖Java5提供的Instrumentation-API，以便于在运行期动态替换或修改类定义，以满足无侵入式需求。在此大家需要注意，关于字节码增强工具的选型，本文采用的是偏向于底层但更具灵活性的ASM框架；当然，如果你本身并不熟悉JVM指令，那么阅读本小节将会显得非常吃力，因此，对于那些不熟悉JVM指令的同学，我建议应该先去了解下常用的JVM指令后再回来阅读本小节。



要想实现协程效果，需要在目标类的2个位置进行插桩。首先是目标方法的开头，其次是在触发“INVOKEVIRTUAL”指令且为挂起函数时。添加前置指令内容，示例1-6：

@Override
public void visitCode() {
    mv.visitCode();
    mv.visitFieldInsn(GETSTATIC, className, "context", "Ljava/lang/ThreadLocal;");
    mv.visitMethodInsn(INVOKEVIRTUAL, "java/lang/ThreadLocal", "get", "()Ljava/lang/Object;", false);
    mv.visitJumpInsn(IFNONNULL, l0);
    mv.visitFieldInsn(GETSTATIC, className, "context", "Ljava/lang/ThreadLocal;");
    mv.visitTypeInsn(NEW, "com/github/coroutine/api/core/Context");
    mv.visitInsn(DUP);
    mv.visitInsn(ICONST_0);
    mv.visitMethodInsn(INVOKESPECIAL, "com/github/coroutine/api/core/Context", "<init>", "(I)V", false);
    mv.visitMethodInsn(INVOKEVIRTUAL, "java/lang/ThreadLocal", "set", "(Ljava/lang/Object;)V", false);
    mv.visitLabel(l0);
    mv.visitFrame(Opcodes.F_SAME, 0, null, 0, null);
    mv.visitFieldInsn(GETSTATIC, className, "context", "Ljava/lang/ThreadLocal;");
    mv.visitMethodInsn(INVOKEVIRTUAL, "java/lang/ThreadLocal", "get", "()Ljava/lang/Object;", false);
    mv.visitTypeInsn(CHECKCAST, "com/github/coroutine/api/core/Context");
    mv.visitVarInsn(ASTORE, 2);
    mv.visitVarInsn(ALOAD, 2);
    mv.visitFieldInsn(GETFIELD, "com/github/coroutine/api/core/Context", "flag", "I");
    mv.visitJumpInsn(IFNE, l1);
}



首先将上下文对象context压入栈顶，通过指令“INVOKEVIRTUAL”调用ThreadLocal.get()实例方法，然后通过指令“IFNONNULL”进行非空验证，如果为null就赋值，反之jump to label0。在label0处，上下文对象context仍然需要经历一遍入/出栈动作，然后执行指令“IFNE”将context.flag字段与字面值0进行匹配，如果不匹配就jump to label1，反之继续执行。这段代码的主要任务是设置初始访问标识，让其进入到特定的代码块中，类似示例1-5中的switch语句。



当触发“INVOKEVIRTUAL”指令且为挂起函数时，需要删除原指令并新增指令，示例1-7：

@Override
public void visitMethodInsn(int opcode, String owner, String name, String desc, boolean itf) {
    //INVOKEVIRTUAL suspend时才继续增强
    if (opcode == Opcodes.INVOKEVIRTUAL && name.equals(Constants.SUSPEND_METHOD_NAME)) {
        mv.visitInsn(Opcodes.POP);
        mv.visitVarInsn(ALOAD, 2);
        mv.visitInsn(ICONST_1);
        mv.visitMethodInsn(INVOKEVIRTUAL, "com/github/coroutine/api/core/Context", "setFlag", "(I)V", false);
        mv.visitVarInsn(ALOAD, 2);
        mv.visitVarInsn(ILOAD, 1);
        mv.visitMethodInsn(INVOKESTATIC, "java/lang/Integer", "valueOf", "(I)Ljava/lang/Integer;", false);
        mv.visitMethodInsn(INVOKEVIRTUAL, "com/github/coroutine/api/core/Context", "setParam", "(Ljava/lang/Object;)V", false);
        mv.visitInsn(RETURN);
        mv.visitLabel(l1);
        mv.visitFrame(Opcodes.F_APPEND, 1, new Object[]{"com/github/coroutine/api/core/Context"}, 0, null);
        mv.visitVarInsn(ALOAD, 2);
        mv.visitFieldInsn(GETFIELD, "com/github/coroutine/api/core/Context", "param", "Ljava/lang/Object;");
        mv.visitTypeInsn(CHECKCAST, "java/lang/Integer");
        mv.visitMethodInsn(INVOKEVIRTUAL, "java/lang/Integer", "intValue", "()I", false);
        mv.visitVarInsn(ISTORE, 1);
        mv.visitFieldInsn(GETSTATIC, className, "context", "Ljava/lang/ThreadLocal;");
        mv.visitMethodInsn(INVOKEVIRTUAL, "java/lang/ThreadLocal", "remove", "()V", false);
        isEnhancement = true;
        return;
    }
    mv.visitMethodInsn(opcode, owner, name, desc, itf);
}



上述指令延续着label0处的逻辑，用于替换指令“INVOKEVIRTUAL suspend”，其主要任务是保存/恢复函数调用堆栈信息，以及根据访问标识控制协程的执行流程。上述示例6~12行代码处，就是函数调用堆栈的保存逻辑，这里的数据结构实现得相对比较简单，上下文对象中只有2个字段，分别为访问标识和方法入参，均通过指令“INVOKEVIRTUAL”调用上下文对象的setter方法进行赋值，随即调用指令“RETURN”返回，类似示例1-5中注释2处的内容。假设协程被切换回来，代码位置14~22行的内容就是在业务逻辑执行前用于reload上下文信息的相关指令，类似示例1-5中注释3处的内容。



至此，本文内容全部结束。如果在阅读过程中有疑问，欢迎加入微信群聊和小伙伴们一起参与讨论。

项目地址

coroutine demo：https://github.com/gaoxianglong/coroutine

码字不易，欢迎转发