python —— 异步geventyeild协程阻塞

Posted 2021-07-27 胖虎是只mao

为什么要用异步？

–这种网络应用通常瓶颈都在IO层面，解决等待读写的问题比提高文本解析速度来的更有性价比。

–而多进程带来的优点（cpu处理）并没有得到体现，反而创建和调度进程带来的开销要远超出它的正面效应，拖了一把后腿。即便如此，多进程带来的效益相比于之前单进程单线程的模型要好得多。（多进程和多线程除了创建的开销大之外还有一个难以根治的缺陷，就是处理进程之间或线程之间的协作问题，因为是依赖多进程和多线程的程序在不加锁的情况下通常是不可控的，而协程则可以完美地解决协作问题，由用户来决定协程之间的调度。）

参考：https://zhuanlan.zhihu.com/p/25228075

一、异步：

异步I / O框架使用非阻塞套接字在单个线程上执行并发操作。Python因为有GIL（全局解释锁）这玩意，不可能有真正的多线程的存在，因此很多情况下都会用multiprocessing实现并发，而且在Python中应用多线程还要注意关键地方的同步，不太方便，用协程代替多线程和多进程是一个很好的选择，因为它吸引人的特性：主动调用/退出，状态保存，避免cpu上下文切换等…

一个生动的例子：
老张爱喝茶，废话不说，煮开水。
出场人物：老张，水壶两把（普通水壶，简称水壶；会响的水壶，简称响水壶）。
1 老张把水壶放到火上，立等水开。（同步阻塞）
老张觉得自己有点傻

2 老张把水壶放到火上，去客厅看电视，时不时去厨房看看水开没有。（同步非阻塞）

3 老张还是觉得自己有点傻，于是变高端了，买了把会响笛的那种水壶。水开之后，能大声发出嘀~~~~的噪音。
老张把响水壶放到火上，立等水开。（异步阻塞）
老张觉得这样傻等意义不大

4 老张把响水壶放到火上，去客厅看电视，水壶响之前不再去看它了，响了再去拿壶。（异步非阻塞）
老张觉得自己聪明了。
所谓同步异步，只是对于水壶而言。
普通水壶，同步；响水壶，异步。
虽然都能干活，但响水壶可以在自己完工之后，提示老张水开了。这是普通水壶所不能及的。
同步只能让调用者去轮询自己（情况2中），造成老张效率的低下。

所谓阻塞非阻塞，仅仅对于老张而言。
立等的老张，阻塞；看电视的老张，非阻塞。
情况1和情况3中老张就是阻塞的，媳妇喊他都不知道。虽然3中响水壶是异步的，可对于立等的老张没有太大的意义。所以一般异步是配合非阻塞使用的，这样才能发挥异步的效用。

参考：https://www.zhihu.com/question/19732473/answer/23434554

==============================================================================================================================================================

什么是阻塞，非阻塞，同步，异步？

一个IO操作其实分成了两个步骤：发起IO请求和实际的IO操作。
阻塞IO和非阻塞IO的区别在于第一步，发起IO请求是否会被阻塞，如果阻塞直到完成那么就是传统的阻塞IO，如果不阻塞，那么就是非阻塞IO。
同步IO和异步IO的区别就在于第二个步骤是否阻塞：如果实际的IO读写阻塞请求进程，那么就是同步IO，因此阻塞IO、非阻塞IO、IO复用、信号驱动IO都是同步IO；如果不阻塞，而是操作系统帮你做完IO操作再将结果返回给你，那么就是异步IO。

二、早期的异步：

用gevent库的猴子补丁，gevent给予了我们一种以同步逻辑来书写异步程序的能力。当我们给程序打了猴子补丁后，Python程序在运行时会动态地将一些网络库（例如socket，thread）替换掉，变成异步的库。使得程序在进行网络操作的时候都变成异步的方式去工作，效率就自然提升很多了。

gevent是第三方库，通过greenlet实现协程，其基本思想是：

当一个greenlet遇到IO操作时，比如访问网络，就自动切换到其他的greenlet，等到IO操作完成，再在适当的时候切换回来继续执行。由于IO操作非常耗时，经常使程序处于等待状态，有了gevent为我们自动切换协程，就保证总有greenlet在运行，而不是等待IO。由于切换是在IO操作时自动完成，所以gevent需要修改Python自带的一些标准库，这一过程在启动时通过monkey patch完成

Python社区也意识到Python需要一个独立的标准库来支持协程，于是就有了后来的asyncio。

三、协程Coroutine：

协程，又称作Coroutine。从字面上来理解，即协同运行的例程，它是比是线程（thread）更细量级的用户态线程，特点是允许用户的主动调用和主动退出，挂起当前的例程然后返回值或去执行其他任务，接着返回原来停下的点继续执行。yield语句实现函数执行到一半返回等会又跑到原来的地方继续执行。（其实这里我们要感谢操作系统（OS）为我们做的工作，因为它具有getcontext和swapcontext这些特性，通过系统调用，我们可以把上下文和状态保存起来，切换到其他的上下文，这些特性为coroutine的实现提供了底层的基础。操作系统的Interrupts和Traps机制则为这种实现提供了可能性）

yeild：
　　yield相当于return，它将相应的值返回给调用next()或者send()的调用者，从而交出了cpu使用权，而当调用者再调用next()或者send()时，又会返回到yield中断的地方，如果send有参数，又会将参数返回给yield赋值的变量，如果没有就跟next()一样赋值为None。

这里先插入生成器的知识：
　　学过生成器和迭代器的同学应该都知道python有yield这个关键字，yield能把一个函数变成一个generator，与return不同，yield在函数中返回值时会保存函数的状态，使下一次调用函数时会从上一次的状态继续执行，即从yield的下一条语句开始执行，这样做有许多好处，比如我们想要生成一个数列，若该数列的存储空间太大，而我们仅仅需要访问前面几个元素，那么yield就派上用场了，它实现了这种一边循环一边计算的机制，节省了存储空间，提高了运行效率。

"""
举例：斐波那契数列
yeild把这个函数变成了生成器，然后当我需要时，调用它的next方法获得下一个值

"""

def fib(max):
    n, a, b = 0, 0, 1
    while n  max:
        yeild b
        a, b = b, a + b
        n = n + 1
        

使用协程的方式解决生产者-消费者模型：

#-*- coding:utf-8
def consumer():
    status = True
    while True:
        n = yield status
        print("我拿到了{}!".format(n))
        if n == 3:
            status = False

def producer(consumer):
    n = 5
    while n > 0:
    # yield给主程序返回消费者的状态
        yield consumer.send(n)    # 生产者调用了消费者的send()方法，把n发送给consumer（即c），在consumer中的n = yield status，n拿到的是消费者发送的数字，同时，consumer用yield的方式把状态（status）返回给消费者，注意：这时producer（即消费者）的consumer.send()调用返回的就是consumer中yield的status！消费者马上将status返回给调度它的主程序，主程序获取状态，判断是否错误，若错误，则终止循环，结束程序。
        n -= 1

if __name__ == '__main__':
    c = consumer()    # 因为consumer函数中存在yield语句，python会把它当成一个generator（生成器，注意：生成器和协程的概念区别很大，千万别混淆了两者），因此在运行这条语句后，python并不会像执行函数一样，而是返回了一个generator object
    c.send(None)   # 将consumer（即变量c，它是一个generator）中的语句推进到第一个yield语句出现的位置，那么在例子中，consumer中的status = True和while True:都已经被执行了，程序停留在n = yield status的位置（注意：此时这条语句还没有被执行），上面说的send(None)语句十分重要，如果漏写这一句，那么程序直接报错
    p = producer(c)  # 像上面一样定义了producer的生成器，注意的是这里我们传入了消费者的生成器，来让producer跟consumer通信
    for status in p:  # 循环地运行producer和获取它yield回来的状态
        if status == False:
            print("我只要3,4,5就行啦")
            break
    print("程序结束")

"""
现在我们要让生产者发送1,2,3,4,5给消费者，消费者接受数字，返回状态给生产者，而我们的消费者只需要3,4,5就行了，当数字等于3时，会返回一个错误的状态。最终我们需要由主程序来监控生产者－消费者的过程状态，调度结束程序
"""

把n发送generator(生成器)中yield的赋值语句中，同时返回generator中yield的变量（结果）。

协程和生成器的区别：
有些人会把生成器（generator）和协程（coroutine）的概念混淆，两者的区别还是很大的。

直接上最重要的区别：

generator总是生成值，一般是迭代的序列

协程关注的是消耗值，是数据(data)的消费者

协程不会与迭代操作关联，而generator会

协程强调协同控制程序流，generator强调保存状态和产生数据

相似的是，它们都是不用return来实现重复调用的函数/对象，都用到了yield(中断/恢复)的方式来实现。

四、asyncio：

asyncio是python 3.4中引入的异步IO库。为简单起见，您需要了解两件事：协程和事件循环。协程就像函数一样，但它们可以在代码中的某些点暂停或恢复。这用于在等待IO（例如HTTP请求）时暂停协程，并在此期间执行另一个协程。我们使用await（等同于yield from）关键字来声明我们想要一个协程的返回值。事件循环用于协调协程的执行。

asyncio是python 3.4中新增的模块，它提供了一种机制，使得你可以用协程（coroutines）、IO复用（multiplexing I/O）在单线程环境中编写并发模型。

根据官方说明，asyncio模块主要包括了：

具有特定系统实现的事件循环（event loop）;

数据通讯和协议抽象（类似Twisted中的部分);

TCP，UDP,SSL，子进程管道，延迟调用和其他;

Future类;

yield from的支持;

同步的支持;

提供向线程池转移作业的接口;

在Python3.5中，引入了aync&await 语法结构，通过"aync def"可以定义一个协程代码片段，作用类似于Python3.4中的@asyncio.coroutine修饰符，而await则相当于"yield from"。

"""
　　当事件循环开始运行时，它会在Task中寻找coroutine来执行调度，因为事件循环注册了print_sum()，因此print_sum()被调用，执行result = await compute(x, y)这条语句（等同于result = yield from compute(x, y)），　　因为compute()自身就是一个coroutine，因此print_sum()这个协程就会暂时被挂起，compute()被加入到事件循环中，程序流执行compute()中的print语句，打印”Compute %s + %s …”，然后执行了await asyncio.sleep(1.0)，　　因为asyncio.sleep()也是一个coroutine，接着compute()就会被挂起，等待计时器读秒，在这1秒的过程中，事件循环会在队列中查询可以被调度的coroutine，而因为此前print_sum()与compute()都被挂起了，因此事件循环会停下来等待协程的调度，　　当计时器读秒结束后，程序流便会返回到compute()中执行return语句，结果会返回到print_sum()中的result中，最后打印result，事件队列中没有可以调度的任务了，此时loop.close()把事件队列关闭，程序结束。

"""

import asyncio

async def compute(x, y):
    print("Compute %s + %s ..." % (x, y))
    await asyncio.sleep(1.0)
    return x + y

async def print_sum(x, y):
    result = await compute(x, y)
    print("%s + %s = %s" % (x, y, result))

loop = asyncio.get_event_loop()
loop.run_until_complete(print_sum(1, 2))
loop.close()

python之gevent

因为python线程的性能问题，在python中使用多线程运行代码经常不能达到预期的效果。而有些时候我们的逻辑中又需要开更高的并发，或者简单的说，就是让我们的代码跑的更快，在同样时间内执行更多的有效逻辑、减少无用的等待。gevent就是一个现在很火、支持也很全面的python第三方协程库。
gevent是python的一个并发框架，以微线程greenlet为核心，使用了epoll事件监听机制以及诸多其他优化而变得高效。而且其中有个monkey类，将现有基于Python线程直接转化为greenlet(类似于打patch)。在运行时的具体流程大概就是：
当一个greenlet遇到IO操作时，比如访问网络/睡眠等待，就自动切换到其他的greenlet，等到IO操作完成，再在适当的时候切换回来继续执行。由于IO操作非常耗时，经常使程序处于等待状态，有了gevent为我们自动切换协程，就保证总有greenlet在运行，而不是等待IO。同时也因为只有一个线程在执行，会极大的减少上下文切换的成本。

gevent内部还是封装的libev异步事件循环。所以，要理解gevent，首先需要对linux的异步I/O有个概念。
举个不恰当的例子，打开一个网页，需要cpu向网卡发送指令，如果是阻塞情况，cpu会一直等待直到网卡通过网络获取到了网页内容，才会继续工作，这样整个工作都是序列化的，并且等待的时间啥都做不了，很显然浪费了cpu。异步的情况则是cpu把指令成功发送给网卡并注册了回调函数以后，就直接返回做其他事情(cpu的利用率上去了，并且程序的执行异步化了)，对应于这个问题，就是cpu向网卡提交了一个网页链接访问请求以后，不等待网卡就继续向网卡提交下一个网页请求(网卡本身也是有控制器的，可以很好的并发处理多个网页请求)。在网卡获取到网页内容以后，通过中断机制告诉cpu，网页获取成功了。cpu接收到通知以后，可以中断当前的执行流程去执行之前注册的回调函数(由于网卡是并发访问网页的，每个网页返回的时间不同导致了回调函数的顺序发生变化)。 不过异步最大的缺陷就是回调太多，程序的执行流程和中间状态保存比较麻烦。而gevent是在这个基础上，以一种同步的方式解决了回调带来的问题。这样更易于编写和维护。

gevent基本使用

# -*- coding: utf-8 -*-

import gevent


def f1():
    for i in range(5):
        print 'run func: f1, index: %s ' % i
        gevent.sleep(0)


def f2():
    for i in range(5):
        print 'run func: f2, index: %s ' % i
        gevent.sleep(0)


t1 = gevent.spawn(f1)
t2 = gevent.spawn(f2)
gevent.joinall([t1, t2])

运行后输出如下图所示：

由图中可以看出，f1和f2是交叉打印信息的，因为在代码执行的过程中，我们人为使用gevent.sleep(0)创建了一个阻塞，gevent在运行到这里时就会自动切换函数切换函数。也可以在执行的时候sleep更长时间，可以发现两个函数基本是同时运行然后各自等待。

在实际运用的过程中，我们如果有需要通过人为sleep来增加时间间隔或者确保部分逻辑安全的时候，此处使用就很方便了。当然，更多时候我们还是在需要进行网络请求的时候使用

gevent :

# -*- coding: utf-8 -*-

from gevent import monkey; monkey.patch_all()
import gevent
import requests
from datetime import datetime


def f(url):
    print 'time: %s, GET: %s' % (datetime.now(), url)
    resp = requests.get(url)
    print 'time: %s, %d bytes received from %s.' % (
        datetime.now(), len(resp.text), url)


gevent.joinall([
        gevent.spawn(f, 'https://www.python.org/'),
        gevent.spawn(f, 'https://www.yahoo.com/'),
        gevent.spawn(f, 'https://github.com/'),
])

运行上述代码，结果如下：

由上图可以看出，程序基本在同一时间触发了对三个网站的请求，然后各自进行，分别结束。也就是当gevent发现阻塞之后，让当前急需执行，然后自动切换到了另外的请求中运行。

加锁
如果需要在使用gevent的时候加锁，也是非常方便的：

# -*- coding: utf-8 -*-

import gevent
from gevent.lock import Semaphore

sem = Semaphore(1)


def f1():
    for i in range(5):
        sem.acquire()
        print 'run f1, this is ', i
        sem.release()
        gevent.sleep(1)


def f2():
    for i in range(5):
        sem.acquire()
        print 'run f2, that is ', i
        sem.release()
        gevent.sleep(0.3)


t1 = gevent.spawn(f1)
t2 = gevent.spawn(f2)
gevent.joinall([t1, t2])

运行结果如下：

由输出可以发现，程序会同时判断是否在sleep以及是否有锁两种情况，然后执行当前的最有操作。

小结
gevent的优势不仅仅是在代码中调用方便，厉害的是它拥有的monkey机制。假设你不愿意修改原来已经写好的python代码，但是又想充分利用gevent机制，那么你就可以用monkey来做到这一点。你所要做的就是在文件开头打一个patch，那么它就会自动替换你原来的thread、socket、time、multiprocessing等代码，全部变成gevent框架。这一切都是由gevent自动完成的。注意这个patch是在所有module都import了之后再打，否则没有效果。
甚至在编写的Web App代码的时候，不需要引入gevent的包，也不需要改任何代码，仅仅在部署的时候，用一个支持gevent的WSGI服务器，就可以获得数倍的性能提升。

本文简单介绍了gevent的使用，下一篇将对gevent的部分源码进行分析。

异步回调

目前为了解决高并发的问题主要有多进程，多线程和协程三种方法。我喜欢用这个例子来解释协程：拿煎牛排为例子，一个牛排一面需要煎三分钟分钟，两面需要煎六分钟。多进程模式：一个厨房（进程）里面有一个厨师（程序），一口锅，一个厨房每6分钟才能生产出一个牛排，n个进程每6分钟能生产n个牛排。多线程模式：一个厨房有x个锅，每个锅都对应一个厨师（线程），也就是一个厨房每6分钟能生产x个牛排，如果采用多进程+多线程，有Y个厨房，每6分钟就能生产x*Y个牛排。多线程的方式貌似解决了高并发的问题，但是，当一个厨师煎牛排一面的3分钟里，只能等着，什么事情也做不了（阻塞），3分钟后翻面，然后又需要等待3分钟才能生产一个牛排。这不是浪费时间浪费资源吗？协程就解决了这个问题，在一个厨房下，放入n口锅，只有一个厨师（单线程），厨师在第一口锅放入牛排后，等待3分钟的过程中可以在其他锅里放入牛排，如果忽略翻面的时间，一个厨房，一个厨师，每6分钟能生产n个牛排。假设翻面+放入牛排耗时1s的话，7分钟能生产牛排60个！！10分钟能生产牛排240个！！！也就是说，制约程序性能的，有可能是等待（阻塞），等待牛排煎熟的过程，等待从硬盘读取数据到内存的过程，等待数据库sql查询的过程，等待www.baidu/com下载的过程，如果网速慢，打开百度需要3s的话，为何不在等待这3s的时间再去打开下一个网页，那个网页先打开，我们就先看那个网页，没必要打开一个看一个。这就是协程的原理。

3. http中用gevent处理异步任务和用celery处理异步任务有什么区别？

celery是将任务丢进一个异步队列，按照处理能力去慢慢处理，比如我想给1000万用户发个推送，丢进celery里面可以开多个任务进程去处理。
gevent是没有新开线程的，适合io操作耗时的任务和io延时大的场景

gevent是在同一个进程中处理， celery是单开进程从消息队列中读取消息，然后执行任务

celery处理海量任务可以很方便使用集群扩容，gevent就局限在web server上了