协程-初探

客官°小女子只卖身不卖艺 2021-10-31 16:10 409阅读 0赞

**目录**

1. 协程定义

2. 进程，线程，协程

3. 协程优点

3.1. 内存小

3.2. 效率高

3.3. 减少同步锁

3.4. 同步代码思维写异步代码

4. 协程的缺点

4.1. 无法利用多核资源

4.2. 协程中不能有阻塞操作

4.3. CPU密集型不是长处

5. Java实现协程（Kilim）的原理

6. 例子

--------------------

# 1. 协程定义 #

Coroutines are computer-program components that generalize subroutines for non-preemptive multitasking, by allowing multiple entry points for suspending and resuming execution at certain locations.

协程是计算机程序组件，用于概括子程序，非抢占式多任务处理，允许多个入口点暂停和，在某些位置恢复执行。

官方的介绍一如既往的专业，下面带大家了解一下什么是协程

# 2. 进程，线程，协程 #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMzMwNjg3_size_16_color_FFFFFF_t_70][]

从图上可以简单看出三个之间的特点。其中线程和协程上一样的，均是独有栈。

我们都知道，进程切换会比线程切换效率慢，是因为切换过程中需要切换的上下文太多。线程只需要切换**栈信息**，而进程需要切换**栈和堆**。（这里的堆是统称所有在堆中的信息，数据，代码，页目录等等）

那么线程与协程之间对比，后者又有什么优势呢？

# 3. 协程优点 #

## 3.1. 内存小 ##

协程只需要极少的栈内存（大概是2～5KB），默认情况下，线程栈的大小为1MB。

**疑问：**协程和线程都是独有栈，为什么协程序占用内存小？

拿Java为例，我们知道，线程是有固定的栈的，基本都是2MB，当然，不同系统可能大小不太一样，但是的确都是固定分配的。这个栈用于保存局部变量。但是如果一个栈只有一个System.out.println("hello world")方法

我们再为此开辟一个2MB的栈大小，似乎有一些大材小用。

因此，在go语言中采用了**动态扩张收缩**的策略：初始化为2KB，最大可扩张到1GB。

## 3.2. 效率高 ##

协程是**用户态线程**。

线程之间的切换受到操作系统的调度。

pc计数器需要从线程A的保存点切换到线程B的上次保存点继续执行，还需要保存线程A的当前上下文。这些事情均由操作系统内核完成。那么将意味着线程切换势必造成系统调用。由**用户态->系统态->用户态**这样两次上下文切换的过程。

协程之间的切换是由虚拟机或者语言层面上的调度器调度

其中的切换是**用户态->用户态→用户态**（中间这个是调度器代码）而且，因为协程的占用内存小，切换所需要保存/恢复上下文也小

还有一个原因在于**线程切换会导致缓存行失效**而协程切换并不会，提高了缓存的命中率。

## 3.3. 减少同步锁 ##

拿生产者消费者举例

如果使用Java实现，需要一个阻塞队列。使用同步锁机制，但如果使用协程将避免使用锁

import time
    
      
    
    def consumer():
    
        r = ''
    
        while True:
    
            n = yield r
    
            if not n:
    
                return
    
            print('[消费者] <--- Consuming %s...' % n)
    
            time.sleep(1)
    
            r = 'ok'
      
    
    def producer(c):
    
        next(c)
    
        n = 0
    
        while n < 5:
    
            n = n + 1
    
            print('[生产者] ---> Producing %s...' % n)
    
            # 和next有同样的同能，可以触发函数到下一个yield，
    
            # 区别于next的是可以像yield的左边的变量传值，例如上面yield左边的n
    
            c_ret = c.send(n) 
    
            print('[生产者] Consumer return %s' % c_ret)
    
        c.close()
    
      
    
    if __name__ == '__main__':
    
        c = consumer()
    
        producer(c)

结果如下

[生产者] ---> Producing 1...
    
    [消费者] <--- Consuming 1...
    
    [生产者] Consumer return ok
    
    [生产者] ---> Producing 2...
    
    [消费者] <--- Consuming 2...
    
    [生产者] Consumer return ok
    
    [生产者] ---> Producing 3...
    
    [消费者] <--- Consuming 3...
    
    [生产者] Consumer return ok
    
    [生产者] ---> Producing 4...
    
    [消费者] <--- Consuming 4...
    
    [生产者] Consumer return ok
    
    [生产者] ---> Producing 5...
    
    [消费者] <--- Consuming 5...
    
    [生产者] Consumer return ok

为什么协程不需要使用锁？因为协程最终还是运行在线程上，本质上还是单线程运行，没有临界区域的话自然不需要锁的机制。多协程自然没有竞争关系。

**但是，如果存在临界区域，依然需要使用锁**

协程可以减少以往必须使用锁的场景

## 3.4. 同步代码思维写异步代码 ##

**异步方式**

#!/usr/bin/python
    
    # -*- coding: UTF-8 -*-
    
    import tornado
    
    from tornado.httpclient import AsyncHTTPClient
    
    import time
    
    import sys
    
      
    
    # 以异步回调方式访问两个url
    
    def http_callback_way(url1, url2):
    
        http_client = AsyncHTTPClient()
    
        begin = time.time()
    
        count = [0]
    
        def handle_result(response, url):
    
            print('%s : handle_result with url %s' % (time.time(), url))
    
            count[0] += 1
    
            if count[0] == 2:
    
                print 'http_callback_way cost', time.time() - begin
    
                sys.exit(0)
    
        # fetch函数有两个参数，第一个是url，第二个是回调函数
    
        # 访问url1
    
        http_client.fetch(url1, lambda res, u=url1: handle_result(res, u))
    
        print('%s here between to request' % time.time())
    
        # 访问url2
    
        http_client.fetch(url2, lambda res, u=url2: handle_result(res, u))
     
    
    url_list = ['https://www.baidu.com', 'https://www.bing.com']
    
    http_callback_way(*url_list)
    
    tornado.ioloop.IOLoop.instance().start()

执行http\_client\_fetch时候，注册回调函数，当请求有返回时候，执行handle\_result，看代码不是线性的看下来的，语义被分割。

**协程方式**

#!/usr/bin/python
    
    # -*- coding: UTF-8 -*-
    
    import tornado
    
    import sys
    
    import time
    
    from tornado.httpclient import AsyncHTTPClient
    
    from tornado import gen
    
      
    
    # 以协程方式访问两个url，使用原生的generator
    
    def http_generator_way(url1, url2):
    
        begin = time.time()
    
        count = [0]
    
      
    
        @gen.coroutine
    
        def do_fetch(url):
    
            http_client = AsyncHTTPClient()
    
            response = yield http_client.fetch(url, raise_error=False)
    
            print url, response.error
    
            count[0] += 1
    
            if count[0] == 2:
    
                print 'http_generator_way cost', time.time() - begin
    
                sys.exit(0)
    
        do_fetch(url1)
    
        do_fetch(url2)
    
      
    
    url_list = ['https://www.baidu.com', 'https://www.bing.com']
    
    http_generator_way(*url_list)
    
    tornado.ioloop.IOLoop.instance().start()

协程的方式实现，执行do\_fetch(url1)，这本身就是一个开启协程通过@gen.coroutine，当yield时候，会主动释放时间片，保存当前协程上下文。那么将继续do\_fetch(url2),然后又执行到yield时候释放时间片，调度到第一个协程哪里，继续往下执行。符合同步逻辑思维

# 4. 协程的缺点 #

## 4.1. 无法利用多核资源 ##

本质上协程还是运行在线程上，单线程应用的话。其实无法利用多核

## 4.2. 协程中不能有阻塞操作 ##

协程是非抢占式，线程是抢占式。

线程在遇见IO操作时候，线程从运行态→阻塞态，释放cpu使用权。这是由操作系统调度。

协程如果遇见IO操作时候，协程是主动释放执行权限的，那么将无法主动释放，程序将阻塞。无法往下执行。随之而来是整个**线程**被阻塞，因为协程运行在线程中。

## 4.3. CPU密集型不是长处 ##

假设这个线程中有一个协程是 CPU 密集型的他没有 IO 操作，也就是自己不会主动触发调度器调度的过程，那么就会出现其他协程得不到执行的情况，所以这种情况下需要程序员自己避免。

在 IO 密集型的程序中由于 IO 操作远远小于 CPU 的操作，所以往往需要 CPU 去等 IO 操作。同步 IO 下系统需要切换线程，让操作系统可以再 IO 过程中执行其他的东西。这样虽然代码是符合人类的思维习惯但是由于大量的线程切换带来了大量的性能的浪费。所以，**使用协程应该是IO密集型**应用。

PS：这里说的IO和4.2不同，4.2是指IO产生的阻塞，其IO发生并没有主动yield切换协程。

# 5. Java实现协程（Kilim）的原理 #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMzMwNjg3_size_16_color_FFFFFF_t_70 1][]

Task：Task.execute相当于于线程Thread.run

Fiber：协程上下文，含有pc计数器，stage堆栈状态

MailBox：协程之间的传递通道

Scheduler：Kilim框架中核心的任务调度器，负责管理任务的工作者线程WorkerThread，以及一个基本的FIFO队列，维护着Task任务列表，Scheduler负责分派Task给指定的工作者线程WorkerThread执行，类似于ExecutorService线程池管理Runnable和Callable任务的执行

任务队列&工作线程队列，字面意思

// original
    
    void a() throws Pasuable {
    
        x = ...
    
        b(); // b is pausable
    
        print (x);
    
    }

只要方法会抛出Pausable，那么字节码将会增强这个方法

// transformed code
    
    void a(Fiber f) {
    
        switch (f.pc) { // prelude
    
            case 0: goto START;
    
            case 1: goto CALL_B}
    
        START:
    
        x = ...
    
        CALL_B: // pre_call
    
        f.down()
    
        b(f);
    
        f.up() // post-call
    
        switch (f.status) {
    
            case NOT_PAUSING_NO_STATE:
    
                goto RESUME
    
            case NOT_PAUSING_HAS_STATE:
    
                restore state
    
                goto RESUME
    
            case PAUSING_NO_STATE :
    
                capture state, return
    
            case PAUSING_HAS_STATE:
    
                return
    
        }
    
        RESUME:
    
        print (x);
    
    }

1.  在prelude块中，也即刚进入函数a时将会执行的代码块，将根据Fiber中的pc程序计数器跳转到对应的代码块处开始执行。
2.  在pre\_call块中，也即在调用函数b之前，将调用Fiber的down方法记录当前执行状态和pc程序计数器，标识着函数将进入下一个Pausable方法。
3.  在post\_call块中，也即在调用函数b之后，将调用Fiber的up方法计算函数b调用完成后返回的状态，标识着从一个被调的Pausable方法返回，它既可能是正常的函数b执行完成返回，也可能是函数b执行暂停返回，接着通过这个状态控制后续的执行流程。

这四种状态分别为：

**NOT\_PAUSING\_\_NO\_STATE**：即被调函数执行完成正常返回，这种情况与即普通的函数执行类似。  
**NOT\_PAUSING\_\_HAS\_STATE**：即被调函数执行完成，但还存在上次暂存的栈帧，这种情况一般是函数从上次暂停处恢复执行，且顺利执行完成返回，此时需要恢复函数的栈帧，然后goto到RESUME代码块继续执行。  
**PAUSING\_\_NO\_STATE**：即被调函数执行过程中暂停，且还未保存函数栈帧，需要主调函数执行暂存操作，这种情况一般即第一次协程执行到需要暂停处，此时需要采用字节码暂存函数的栈帧和状态，然后直接return。  
**PAUSING\_\_HAS\_STATE**：即被调函数执行过程中暂停，且已经保存函数栈帧，这种情况是该Pausable从上次暂停处恢复执行，但是依然没有预期的结果，需要再次暂停，此时因为之前暂停时函数栈帧和状态都已经保存过，不需要再做什么，因此直接return即可。

# 6. 例子 #

public class Main {
    
    
    
        public static void main(String[] args) throws InterruptedException {
    
            Task1 task1 = new Task1();
    
            Task2 task2 = new Task2();
    
            task1.start();
    
            task2.start();
    
            task1.resume();
    
            task2.resume();
    
            Thread.sleep(1000);
    
        }
    
    
    
    
    
    }
    
    
    
    class Task1 extends Task {
    
        @Override
    
        public void execute() throws Pausable {
    
            System.out.println("wxt");
    
            yield();   //PAUSING__NO_STATE
    
            System.out.println("biu biu biu");
    
            System.out.println("999");
    
        }
    
    }
    
    
    
    class Task2 extends Task {
    
        @Override
    
        public void execute() throws Pausable {
    
            System.out.println("taotao");
    
            yield();  //PAUSING__NO_STATE
    
            System.out.println("666");
    
        }
    
    }

输出

wxt
    
    taotao
    
    biu biu biu
    
    999
    
    666

class Task1 extends Task {
    
        public static final boolean $isWoven = true;
    
        Task1() {
    
        }
    
        public void execute(Fiber var1) throws Pausable {
    
            switch(var1.pc) {
    
            case 0:
    
                System.out.println("wxt");
    
            case 1:
    
                break;
    
            default:
    
                var1.wrongPC();
    
            }
    
            yield(var1.down());
    
            switch(var1.up()) {
    
            case 0:  //NOT_PAUSING__NO_STATE
    
            case 1: //NOT_PAUSING__HAS_STATE
    
            default:
    
                System.out.println("biu biu biu");
    
                System.out.println("999");
    
                return;
    
            case 2: //PAUSING__NO_STATE
    
                State var2 = new State();
    
                var2.self = this;
    
                var2.pc = 1;
    
                var1.setState(var2);
    
                return;
    
            case 3: //PAUSING__HAS_STATE
    
            }
    
        }
    
    
        public void execute() throws Pausable {
    
            Task.errNotWoven();
    
        }
    
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMzMwNjg3_size_16_color_FFFFFF_t_70]: /images/20211031/4370612a14994830b1fa98cace098d63.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMzMwNjg3_size_16_color_FFFFFF_t_70 1]: /images/20211031/04ab692520b94b08814481c1dbb1ae96.png