Java之协程（quasar）

　1、前面咱们简单的说了一下，Python中的协程原理。这里补充Java的协程实现过程。有须要能够查看python之协程。

　　2、Java协程，其实作Java这么久我也没有怎么听过Java协程的东西，可是一直有有听到微线程/协程的概念，这不在学习Python的时候接触到了协程一词。而后返回来去了解Java的协程问题，可是看了不少资料，发现官网以及不少地方都没有涉及到协程的东西，没有办法，只能经过强大的社区来学习协程的相关东西。

　　3、这里主要关注的是：quasar。

　　1）协程的目的：当咱们在使用多线程的时候，若是存在长时间的I/O操做。这个时候线程一直处于阻塞状态，若是线程不少的时候，会存在不少线程处于空闲状态，形成了资源应用不完全。相对的协程不同了，在单线程中多个任务来回自行若是出现长时间的I/O操做，让其让出目前的协程调度，执行下一个任务。固然可能全部任务，所有卡在同一个点上，可是这只是针对于单线程而言，当全部数据正常返回时，会同时处理当前的I/O操做。

　　2）多线程测试（这里使用100万个线程，来测试内存占用）　

for (int i = 0; i < 1000000; i++) {
            new Thread(() -> {
                try {
                    Thread.sleep(100000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
        }

　　结果：

　　直接卡死了，内存溢出

　　

　　可想而知，若是存在100万个线程，开销是有多大。

　　3）协程测试

　　a、阿里云搜索到的依赖包

　　　　<dependency>
            <groupId>co.paralleluniverse</groupId>
            <artifactId>quasar-core</artifactId>
            <version>0.7.9</version>
            <classifier>jdk8</classifier>
        </dependency>

　　b、测试内存占用量

　　public static void main(String[] args) throws Exception {
        //使用阻塞队列来获取结果。
        LinkedBlockingQueue<Fiber<Integer>> fiberQueue = new LinkedBlockingQueue<>();
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        for (int i = 0; i < 1000000; i++) {
            int finalI = i;
            //这里的Fiber有点像Callable,能够返回数据
            Fiber<Integer> fiber = new Fiber<>((SuspendableCallable<Integer>) () -> {
                //这里用于测试内存占用量
                Fiber.sleep(100000);
                System.out.println("in-" + finalI + "-" + LocalDateTime.now().format(formatter));
                return finalI;
            });
            //开始执行
            fiber.start();
            //加入队列
            fiberQueue.add(fiber);
        }
        while (true) {
            //阻塞
            Fiber<Integer> fiber = fiberQueue.take();
            System.out.println("out-" + fiber.get() + "-" + LocalDateTime.now().format(formatter));
        }
    }

　　结果：

　　堆：

　　

 　　估计：1个G左右。

　　内存：

　　

　　估计：1个G左右，也就是每个fiber占用1Kb左右。

 　　c、正常测试

　　修改一下参数：

public static void main(String[] args) throws Exception {
        //使用阻塞队列来获取结果。
        LinkedBlockingQueue<Fiber<Integer>> fiberQueue = new LinkedBlockingQueue<>();
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        for (int i = 0; i < 10; i++) {
            int finalI = i;
            //这里的Fiber有点像Callable,能够返回数据
            Fiber<Integer> fiber = new Fiber<>((SuspendableCallable<Integer>) () -> {
                //这里用于测试内存占用量
                Fiber.sleep(1000);
                System.out.println("in-" + finalI + "-" + LocalDateTime.now().format(formatter));
                return finalI;
            });
            //开始执行
            fiber.start();
            //加入队列
            fiberQueue.add(fiber);
        }
        while (true) {
            //阻塞
            Fiber<Integer> fiber = fiberQueue.take();
            System.out.println("out-" + fiber.get() + "-" + LocalDateTime.now().format(formatter));
        }
    }

　　结果：

　　

　　4）能够看出并发的状态仍是很不错的，固然这只是多个任务执行而已。

　　4、经过上面的测试，能够看出，quasar中Fiber，很像Callable的用法、并且在内存占用上面减小了不少，固然，堆的数量确实很多，可是能够接受。

　　仍是要说明：协程的方式更多用来作I/O密集型的操做。计算密集型的仍是使用线程更加合理。

　　5、原理，我估么着看了一下源码，复杂度很高，这里不作深究。

　　原理参考：次时代Java编程（一）：Java里的协程

　　一、Quasar里的Fiber实际上是一个continuation，他能够被Quasar定义的scheduler调度，一个continuation记录着运行实例的状态，并且会被随时中断，而且也会随后在他被中断的地方恢复。Quasar实际上是经过修改bytecode来达到这个目的，因此运行Quasar程序的时候，你须要先经过java-agent在运行时修改你的代码，固然也能够在编译期间这么干。golang的内置了本身的调度器，Quasar则默认使用ForkJoinPool这个JDK7之后才有的，具备work-stealing功能的线程池来当调度器。work-stealing很是重要，由于你不清楚哪一个Fiber会先执行完，而work-stealing能够动态的从其余的等等队列偷一个context过来，这样能够最大化使用CPU资源。

　　二、那这里你会问了，Quasar怎么知道修改哪些字节码呢，其实也很简单，Quasar会经过java-agent在运行时扫描哪些方法是能够中断的，同时会在方法被调用前和调度后的方法内插入一些continuation逻辑，若是你在方法上定义了@Suspendable注解，那Quasar会对调用该注解的方法作相似下面的事情。

　　三、这里假设你在方法f上定义了@Suspendable，同时去调用了有一样注解的方法g，那么全部调用f的方法会插入一些字节码，这些字节码的逻辑就是记录当前Fiber栈上的状态，以便在将来能够动态的恢复。(Fiber相似线程也有本身的栈)。在suspendable方法链内Fiber的父类会调用Fiber.park，这样会抛出SuspendExecution异常，从而来中止线程的运行，好让Quasar的调度器执行调度。这里的SuspendExecution会被Fiber本身捕获，业务层面上不该该捕获到。若是Fiber被唤醒了(调度器层面会去调用Fiber.unpark)，那么f会在被中断的地方从新被调用(这里Fiber会知道本身在哪里被中断)，同时会把g的调用结果(g会return结果)插入到f的恢复点，这样看上去就好像g的return是f的local variables了，从而避免了callback嵌套。

　　四、上面啰嗦了一大堆，其实简单点讲就是，想办法让运行中的线程栈停下来，好让Quasar的调度器介入。JVM线程中断的条件只有两个，一个是抛异常，另一个就是return。这里Quasar就是经过抛异常的方式来达到的，因此你会看到我上面的代码会抛出SuspendExecution。可是若是你真捕获到这个异常，那就说明有问题了，因此通常会这么写。