day10--协成异步IO缓存

   协成(Gevent)

    协程,又称微线程,纤程。英文名Coroutine。一句话说明什么是线程:协程是一种用户态的轻量级线程。CPU只认识线程。

    协程拥有自己的寄存器上下文和栈。协程调度切换时,将寄存器上下文和栈保存到其他地方,在切回来的时候,恢复先前保存的寄存器上下文和栈。因此:

    协程能保留上一次调用时的状态(即所有局部状态的一个特定组合),每次过程重入时,就相当于进入上一次调用的状态,换种说法:进入上一次离开时所处逻辑流的位置。

    协程的好处:

    1.无需线程上下文切换的开销;

    2.无需原子操作锁定及同步的开销;

    "原子操作(atomic operation)是不需要synchronized",所谓原子操作是指不会被线程调度机制打断的操作;这种操作一旦开始,就一直运行到结束,中间不会有任何 context switch (切换到另一个线程)。原子操作可以是一个步骤,也可以是多个操作步骤,但是其顺序是不可以被打乱,或者切割掉只执行部分。视作整体是原子性的核心。改变量即可称为简单原子操作。协成是在单线程里面实现的。同一时间只有一个线程。

    3.方便切换控制流,简化编程模型;

    4.高并发+高扩展性+低成本:一个CPU支持上万的协程都不是问题。所以很适合用于高并发处理。

    缺点:

    1.无法利用多核资源:协程的本质是个单线程,它不能同时将单个CPU的多个核用上,协程需要和进程配合才能运行在多CPU上.当然我们日常所编写的绝大部分应用都没有这个必要,除非是cpu密集型应用;Ngix只有一个进程,一个进程里面只有一个线程。

    2.进行阻塞(Blocking)操作(如IO时)会阻塞掉整个程序;

    使用yield实现协程操作例子

import time
import queue

def consumer(name):
    print("--->starting eating baozi...")
    while True:
        new_baozi = yield
        print("[%s] is eating baozi %s" % (name, new_baozi))
        # time.sleep(1)

def producer():
    r = con.__next__()
    r = con2.__next__()
    n = 0
    while n < 5:
        n += 1
        con.send(n)
        con2.send(n)
        print("33[32;1m[producer]33[0m is making baozi %s" % n)

if __name__ == '__main__':
    con = consumer("c1")
    con2 = consumer("c2")
    p = producer()

    看楼上的例子,我问你这算不算做是协程呢?你说,我他妈哪知道呀,你前面说了一堆废话,但是并没告诉我协程的标准形态呀,我腚眼一想,觉得你说也对,那好,我们先给协程一个标准定义,即符合什么条件就能称之为协程:

    1.必须在只有一个单线程里实现并发;

    2.修改共享数据不需加锁;

    3.用户程序自己保存多个控制流的上下文栈;

    4.一个协成遇到IO操作自动切换到其它协成。

    基于上面这4点定义,我们刚才用yield实现的程并不能算是合格的线程,因为它有一点功能没实现,哪一点呢?

    Greenlet

    greenlet是一个用C实现的协程模块,相比与python自带的yield,它可以使你在任意函数之间随意切换,而不需把这个函数先声明为generator

from greenlet import greenlet

def test1():
    print(12)
    gr2.switch()
    print(34)
    gr2.switch()

def test2():
    print(56)
    gr1.switch()
    print(78)

gr1 = greenlet(test1)                 #启动一个协成
gr2 = greenlet(test2)
gr1.switch()                          #切换,手动切换

    上面代码执行结果如下:

12
56
34
78

    上面代码中,greenlet模块,greenlet类中,主要实现不同方法之间的切换,让程序能上下文进行切换,switch()转换gr1.switch()

    协成是遇到IO操作进行切换。做事需要花费时间过长,比如从数据库获取数据等等。协成也是串行的,只是在IO操作之间进行切换。来回在IO操作之间切换。

    Gevent是自动切换,封装了greenlet,greenlet还是手动切换,通过switch()手动切换。而Gevent就是自动切换。   

   Gevent

    Gevent 是一个第三方库,可以轻松通过gevent实现并发同步或异步编程,在gevent中用到的主要模式是Greenlet, 它是以C扩展模块形式接入Python的轻量级协程。 Greenlet全部运行在主程序操作系统进程的内部,但它们被协作式地调度。   

    下面我们来看一个协成的实例,如下:

'''协成:微线程,主要作用是遇到IO操作就切换,程序本身就串行的,主要是在各个IO直接来回切换,节省时间'''
import gevent

def fun():
    print("In the function fun!!!
")                                    #(1)
    gevent.sleep(2)                                                      #IO阻塞,也可以是具体执行一个操作,这里直接等待
    print("33[31mCome back to the function of fun again!33[0m
")    #(6)

def inner():
    '''定义一个函数,包含IO阻塞'''
    print("Running in the function of inner!!!
")                        #(2)
    gevent.sleep(1)                                                       #触发切换,遇到gevent.sleep()触发切换,执行后面代码。
    print("33[32mRunning back to the function of inner33[0m
")       #(5)

def outer():
    '''定义一个函数,也包含IO阻塞'''
    print("周末去搞基把,带上%s
" %"alex")                                 #(3)
    gevent.sleep(0.5)
    print("不好吧,%s是人妖,不好那口,还是带上配齐把!!!"  %"alex")          #(4)

gevent.joinall([
    gevent.spawn(fun),
    gevent.spawn(inner),
    gevent.spawn(outer),
])

    上面代码执行结果如下:

In the function fun!!!

Running in the function of inner!!!

周末去搞基把,带上alex

不好吧,alex是人妖,不好那口,还是带上配齐把!!!
Running back to the function of inner

Come back to the function of fun again!

   上面程序中,使用的是协成,我们可以看出,协成遇到IO操作是阻塞的,协成是在串行情况下实现了多并发或异步的效果,整个程序串行执行的话需要至少花费3.5秒,而使用协成花费:2.0312681198120117,可见如果有IO阻塞,协成能够有效降低程序运行的时间。

    gevent.spawn(outer,"alex","wupeiqi"),gevent.spawn()里面加参数的情况,第一个是函数名,后面加引号是参数。让协成等待的方法是gevent.sleep(time),让协成等待,gevent.sleep(),而不是time.sleep(),如果是time.sleep()是不会切换的,gevent.sleep()才会切换。如果是time.sleep(2)秒,是不会切换的,程序将变成串行。执行最长的IO。

    如何判断是IO操作,这个很重要,因为我在自己尝试的时候,time.sleep(2)是不算IO操作的。不会发生切换。

    上面我们使用gevent.sleep()来判断IO操作切换,使用time.sleep()是不行的,程序还是串行的,如何让程序知道time.sleep()是IO操作呢?要打上一个补丁,from gevent import monkey  并且声明:monkey.patch_all(),这样程序就会自动检测IO操作,程序就变成串行的了,如下所示:

'''协成:微线程,主要作用是遇到IO操作就切换,程序本身就串行的,主要是在各个IO直接来回切换,节省时间'''
import gevent,time
from gevent import monkey

monkey.patch_all()                                                     #把当前程序的所有的IO操作给单独的做上标记
def fun():
    print("In the function fun!!!
")
    time.sleep(2)                                                      #IO阻塞,也可以是具体执行一个操作,这里直接等待
    print("33[31mCome back to the function of fun again!33[0m
")

def inner():
    '''定义一个函数,包含IO阻塞'''
    print("Running in the function of inner!!!
")
    time.sleep(1)
    print("33[32mRunning back to the function of inner33[0m
")

def outer(name,arg):
    '''定义一个函数,也包含IO阻塞'''
    print("周末去搞基把,带上%s
" %name)
    time.sleep(0.5)
    print("不好吧,%s是人妖,不好那口,还是带上%s!!!"  %(name,arg))

start_time = time.time()
gevent.joinall([
    gevent.spawn(fun),
    gevent.spawn(inner),
    gevent.spawn(outer,"alex","wupeiqi"),
])
end_time = time.time()
print("花费时间:%s" %(end_time -start_time))

    上面程序的执行结果如下:

In the function fun!!!

Running in the function of inner!!!

周末去搞基把,带上alex

不好吧,alex是人妖,不好那口,还是带上wupeiqi!!!
Running back to the function of inner

Come back to the function of fun again!

花费时间:2.0013585090637207

    可见,打上补丁之后,from gevent import monkey,声明monkey.patch_all()就能让程序自动检测哪些是IO操作,不需要自己注明。

    下面,我们用协成来简单爬取一下网页,如下:

import gevent,time
from urllib.request import urlopen
'''导入urllib模块中的urlopen用来打开网页'''

def func(url):
    print("33[31m----------------------并发的爬取未网页---------------------------33[0m")
    resp = urlopen(url)                                      #使用urlopen打开一个网址
    data = resp.read()                                       #读取网页里面的内容
    with open("alex.html",'wb') as f:
        f.write(data)
    print("%s个字节被下载,从网页%s。" %(len(data),url))

# if __name__ == "__mian__":
start_time = time.time()
gevent.joinall([
    gevent.spawn(func,"https://www.cnblogs.com/alex3714/articles/5248247.html"),
    gevent.spawn(func,"https://www.python.org/"),
    gevent.spawn(func,"https://www.jd.com/")
    ])
end_time = time.time()
cost_time = end_time - start_time
print("cost time:%s" %cost_time)

    执行结果如下:

----------------------并发的爬取未网页---------------------------
91829个字节被下载,从网页https://www.cnblogs.com/alex3714/articles/5248247.html。
----------------------并发的爬取未网页---------------------------
48721个字节被下载,从网页https://www.python.org/。
----------------------并发的爬取未网页---------------------------
124072个字节被下载,从网页https://www.jd.com/。
cost time:18.932090044021606

    从上面结果可以看出,程序肯定是串行执行的,因为是先爬取第一个网页,爬取完了才爬取第二个网页,异步肯定不是这样,为什么会这样呢?因为urllib是被检测不到进行了urllib的,要想gevent检测到urllib,必须进行声明,让gevent能够自动检测到IO操作,from gevent import monkey,mokdey.patch_all()(把当前程序的所有的IO操作给我单独的做上标记,让程序检测到所有有可能进行IO操作的,打一个标记)。下面我们来给上面程序加上标记:

    遇到IO阻塞时会自动切换任务

import gevent,time
from urllib.request import urlopen
from gevent import monkey
monkey.patch_all()
'''导入urllib模块中的urlopen用来打开网页'''

def func(url):
    print("33[31m----------------------并发的爬取未网页---------------------------33[0m")
    resp = urlopen(url)                                      #使用urlopen打开一个网址
    data = resp.read()                                       #读取网页里面的内容
    with open("alex.html",'wb') as f:
        f.write(data)
    print("%s个字节被下载,从网页%s。" %(len(data),url))

# if __name__ == "__mian__":
start_time = time.time()
gevent.joinall([
    gevent.spawn(func,"https://www.cnblogs.com/alex3714/articles/5248247.html"),
    gevent.spawn(func,"https://www.python.org/"),
    gevent.spawn(func,"https://www.jd.com/")
    ])
end_time = time.time()
cost_time = end_time - start_time
print("cost time:%s" %cost_time)

    代码执行结果如下:

----------------------并发的爬取未网页---------------------------
----------------------并发的爬取未网页---------------------------
----------------------并发的爬取未网页---------------------------
124072个字节被下载,从网页https://www.jd.com/。
91829个字节被下载,从网页https://www.cnblogs.com/alex3714/articles/5248247.html。
48721个字节被下载,从网页https://www.python.org/。
cost time:1.7704188823699951

    可以看出,给程序加上能够自动识别IO操作的标记之后,爬虫的时间少了很多,串行需要18秒,现在只需不到2秒,节约了很多时间,这就是异步的效率。

    通过gevent实现单线程下的多socket并发

    下面我们用协成来写一个socket多并发的实例,如下:

    server

import sys
import socket
import time
import gevent
 
from gevent import socket,monkey
monkey.patch_all()
 
 
def server(port):
    s = socket.socket()
    s.bind(('0.0.0.0', port))
    s.listen(500)
    while True:
        cli, addr = s.accept()
        gevent.spawn(handle_request, cli)                    #在连接之后启动一个协成,socket中启动的是一个线程。
 
 
 
def handle_request(conn):
    try:
        while True:
            data = conn.recv(1024)
            print("recv:", data)
            conn.send(data)
            if not data:
                conn.shutdown(socket.SHUT_WR)
 
    except Exception as  ex:
        print(ex)
    finally:
        conn.close()
if __name__ == '__main__':
    server(8001)

    Client

import socket
 
HOST = 'localhost'    # The remote host
PORT = 8001           # The same port as used by the server
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect((HOST, PORT))
while True:
    msg = bytes(input(">>:"),encoding="utf8")
    s.sendall(msg)
    data = s.recv(1024)
    #print(data)
 
    print('Received', repr(data))
s.close()

    其实使用协成就能实现异步的操作,在socket中,正常要想使用多并发,要使用socketserver,但是这里我们使用协成就能实现多并发的情况,gevent.spawn(function,parameter)

    协成是如何切换会IO操作之前的位置。

    论事件驱动与异步IO

    通常,我们写服务器处理模型的程序时,有以下几种模型:

    (1)每收到一个请求,创建一个新的进程,来处理该请求;

    (2)每收到一个请求,创建一个新的线程,来处理该请求;

    (3)每收到一个请求,放入一个事件列表,让主进程通过非阻塞I/O方式来处理请求;(以事件驱动的模式)

    上面的几种方式,各有千秋,

    第(1)中方法,由于创建新的进程的开销比较大,所以,会导致服务器性能比较差,但实现比较简单。

    第(2)种方式,由于要涉及到线程的同步,有可能会面临死锁等问题。

    第(3)种方式,在写应用程序代码时,逻辑比前面两种都复杂。

    综合考虑各方面因素,一般普遍认为第(3)种方式是大多数网络服务器采用的方式

    看图说话讲事件驱动模型

    在UI编程中,常常要对鼠标点击进行相应,首先如何获得鼠标点击呢?

    方式一:创建一个线程,该线程一直循环检测是否有鼠标点击,那么这个方式有以下几个缺点

    1. CPU资源浪费,可能鼠标点击的频率非常小,但是扫描线程还是会一直循环检测,这会造成很多的CPU资源浪费;如果扫描鼠标点击的接口是阻塞的呢?

    2. 如果是堵塞的,又会出现下面这样的问题,如果我们不但要扫描鼠标点击,还要扫描键盘是否按下,由于扫描鼠标时被堵塞了,那么可能永远不会去扫描键盘;

    3. 如果一个循环需要扫描的设备非常多,这又会引来响应时间的问题;

    所以,该方式是非常不好的。

    方式二:就是事件驱动模型

    目前大部分的UI编程都是事件驱动模型,如很多UI平台都会提供onClick()事件,这个事件就代表鼠标按下事件。事件驱动模型大体思路如下:

    1. 有一个事件(消息)队列;

    2. 鼠标按下时,往这个队列中增加一个点击事件(消息);

    3. 有个循环,不断从队列取出事件,根据不同的事件,调用不同的函数,如onClick()、onKeyDown()等;

    4. 事件(消息)一般都各自保存各自的处理函数指针,这样,每个消息都有独立的处理函数;

    事件驱动编程是一种编程范式,这里程序的执行流由外部事件来决定。它的特点是包含一个事件循环,当外部事件发生时使用回调机制来触发相应的处理。另外另种常见编程范式是(单线程)同步以及多线程编程。

    让我们用例子来比较和对比一下单线程、多线程以及事件驱动编程模型。下图展示了随着时间的推移,这三种模式下程序所做的工作。这个程序有3个任务需要完成,每个任务都在等待I/O操作时阻塞自身。阻塞在I/O操作上所花费的时间已经用灰色框标示出来了。

    上图中,灰色部分是IO阻塞。

    在单线程同步模型中,任务按照顺序执行。如果某个任务因为I/O而阻塞,其他所有的任务都必须等待,直到它完成之后它们才能依次执行。这种明确的执行顺序和串行化处理的行为是很容易推断得出的。如果任务之间并没有互相依赖的关系,但仍然需要互相等待的话这就使得程序不必要的降低了运行速度。

    在多线程版本中,这3个任务分别在独立的线程中执行。这些线程由操作系统来管理,在多处理器系统上可以并行处理,或者在单处理器系统上交错执行。这使得当某个线程阻塞在某个资源的同时其他线程得以继续执行。与完成类似功能的同步程序相比,这种方式更有效率,但程序员必须写代码来保护共享资源,防止其被多个线程同时访问。多线程程序更加难以推断,因为这类程序不得不通过线程同步机制如锁、可重入函数、线程局部存储或者其他机制来处理线程安全问题,如果实现不当就会导致出现微妙且令人痛不欲生的bug。

    在事件驱动版本的程序中,3个任务交错执行,但仍然在一个单独的线程控制中。当处理I/O或者其他昂贵的操作时,注册一个回调到事件循环中,然后当I/O操作完成时继续执行。回调描述了该如何处理某个事件。事件循环轮询所有的事件,当事件到来时将它们分配给等待处理事件的回调函数。这种方式让程序尽可能的得以执行而不需要用到额外的线程。事件驱动型程序比多线程程序更容易推断出行为,因为程序员不需要关心线程安全问题。

    IO操作识别是通过事件驱动来实现的,一遇到IO操作。

当我们面对如下的环境时,事件驱动模型通常是一个好的选择:

  1. 程序中有许多任务,而且…
  2. 任务之间高度独立(因此它们不需要互相通信,或者等待彼此)而且…
  3. 在等待事件到来时,某些任务会阻塞。

当应用程序需要在任务间共享可变的数据时,这也是一个不错的选择,因为这里不需要采用同步处理。

网络应用程序通常都有上述这些特点,这使得它们能够很好的契合事件驱动编程模型。

    此处要提出一个问题,就是,上面的事件驱动模型中,只要一遇到IO就注册一个事件,然后主程序就可以继续干其它的事情了,只到io处理完毕后,继续恢复之前中断的任务,这本质上是怎么实现的呢?哈哈,下面我们就来一起揭开这神秘的面纱。。。。

    SelectPollEpoll异步IO 

  http://www.cnblogs.com/alex3714/p/4372426.html 

  番外篇 http://www.cnblogs.com/alex3714/articles/5876749.html

    IO切换实在内核里面进行。必须要copy()一下。

    二 IO模式

  刚才说了,对于一次IO访问(以read举例),数据会先被拷贝到操作系统内核的缓冲区中,然后才会从操作系统内核的缓冲区拷贝到应用程序的地址空间。所以说,当一个read操作发生时,它会经历两个阶段:
  1. 等待数据准备 (Waiting for the data to be ready)
  2. 将数据从内核拷贝到进程中 (Copying the data from the kernel to the process)

  正式因为这两个阶段,linux系统产生了下面五种网络模式的方案。
  - 阻塞 I/O(blocking IO)
  - 非阻塞 I/O(nonblocking IO)
  - I/O 多路复用( IO multiplexing)
  - 信号驱动 I/O( signal driven IO)
  - 异步 I/O(asynchronous IO)

  注:由于signal driven IO在实际中并不常用,所以我这只提及剩下的四种IO Model。

    非阻塞(nonblocking IO)的特点是用户进程需要不断的主动询问kernel数据好了没有。

    I/O 多路复用(IO multiplexing)就是我们说的select,poll,epoll,有些地方也称这种IO方式为event driven IO。select/epoll的好处就在于单个process就可以同时处理多个网络连接的IO。它的基本原理就是select,poll,epoll这个function会不断的轮询所负责的所有socket,当某个socket有数据到达了,就通知用户进程。

    server.setblocking(0)是设置socket非阻塞,默认是server.setblocking(True)阻塞状态。select.select(inputs,outputs,inputs)用来监听链接。select.select()用来监听,可以监听自己。inputs=[server,];readable,writeable,exceptional = select.select(inputs,outputs,inputs);

    select写的多并发:

    Server

'''用select写客户端和socket是差不多的,只是select用的非阻塞的IO多路复用,要设置为非阻塞状态'''
import socket,select
#select是用来监听那个链接处于活跃状态,如果是Server说明是需要建立一个新的连接,如果是conn,说明conn链接处于活动状态,可以收发数据

server = socket.socket()                                           #设置为非阻塞状态
IP,PORT = "0.0.0.0",9999
server.bind((IP,PORT))
server.listen()
server.setblocking(False)
inputs = [server,]                                                        #定义一个列表,用来放置内核检测的链接
#inputs=[server,conn]有新的链接来了就放进去,让select来监测
outputs = []
'''如果有链接进来,会返回三个数据'''
flag = True
while flag:            #死循环,让程序一直监听是否有新的链接过来,如果有活动
    readable,writeable,exceptional = select.select(inputs,outputs,inputs)                               #内核用来监听连接,没有监听内容就报错,监听server
    '''outputs,exceptional是放置断的链接,监听那些连接失效了,没有链接就监听自己,当活动的时候就代表有新的连接进来了'''
    print(readable,writeable,exceptional)
    '''[<socket.socket fd=3, family=AddressFamily.AF_INET, type=2049, proto=0, laddr=('0.0.0.0', 9998)>] [] []链接后的结果'''
    '''连接上来之后,客户端就挂断了'''
    for r in readable:
        if r is server:   #如果来了一个新连接,需要创建新链接
            conn,addr = server.accept()                                           #没有链接进来不要accept()因为是非阻塞的
            print(conn,addr)
            inputs.append(conn)     #因为这个新建立的连接还没有发数据过来,现在就接收的话程序就报错了
            '''所以要想实现这个客户端发数据来时server端能知道,就需要让select在监测这个conn,生成一个连接select就监测'''
        else:
            data = r.recv(1024)
            if not data:
                flag = False
            print("收到数据",data.decode("utf-8"))
            r.send(data.upper())

     Client(客户端)

import socket
client = socket.socket()
ip,port = 'localhost',9999
client.connect((ip,port))
while True:
    mess = input(">>:").encode("utf-8").strip()
    if len(mess) == 0:
        print("发送消息不能为空!!!")
        continue
    client.send(mess)
    data = client.recv(1024)
    print("客户端接收返回数据:%s" %data.decode("utf-8"))

    select能够实现多并发的原因是select是用来监听线程的活跃状态,如果有活跃的线程,select能够监测到,那个线程处于活跃状态,就会监测那个线程,并进行数据的交换。

    readable,writeable,exceptional = select.select(inputs,outputs,inputs)用来生成几种状态,监听的连接放在Inputs中。

    for e in exceptional:
    inputs.remove(e)
    上面程序中,如果客户端断开,则会发生死循环,如何解决呢?就在select.select()中的exceptional,我们要删除活跃的链接,inputs.remove(3),就是删除活跃的链接。

    selectors模块

    This module allows high-level and efficient I/O multiplexing, built upon the select module primitives. Users are encouraged to use this module instead, unless they want precise control over the OS-level primitives used.

    selectors模块默认是使用epoll(),如果系统不支持(Windows)则使用select()方法。

import selectors
import socket
 
sel = selectors.DefaultSelector()
 
def accept(sock, mask):
    conn, addr = sock.accept()  # Should be ready
    print('accepted', conn, 'from', addr)
    conn.setblocking(False)
    sel.register(conn, selectors.EVENT_READ, read)
 
def read(conn, mask):
    data = conn.recv(1000)  # Should be ready
    if data:
        print('echoing', repr(data), 'to', conn)
        conn.send(data)  # Hope it won't block
    else:
        print('closing', conn)
        sel.unregister(conn)
        conn.close()
 
sock = socket.socket()
sock.bind(('localhost', 10000))
sock.listen(100)
sock.setblocking(False)
sel.register(sock, selectors.EVENT_READ, accept)
 
while True:
    events = sel.select()
    for key, mask in events:
        callback = key.data
        callback(key.fileobj, mask)

    使用selectors模块编写的语句,epoll()方法。

原文地址:https://www.cnblogs.com/gengcx/p/7502813.html