并发编程(三) 线程池进程池,协程

1.进程池和线程池

　　开进程和开线程都需要消耗资源,只不过线程相比进程耗费的资源较小,但是计算机的硬件是有限制的,我们不能无限制的去开启进程或者线程.进程池和线程池能帮助我们在计算机承受的范围内最大限度的利用计算机

什么是池

　　在保证计算机硬件安全的情况下最大限度的利用计算机

　　池其实是降低了程序的运行效率,但是保证了计算机硬件的安全

怎么使用池

　　我们首先定义一个池子,在里边放入固定数量的进程或线程,只要有任务来了,就派一个进程或线程去处理任务,如果固定数量的进程或线程使用完了,那么接下来的任务就在外面等待,直到有进程或线程的任务执行完毕,外面的任务才可以拿到空闲的进程或线程执行,这样我们既可以同时使用多个进程或线程,又可以保证计算机硬件不会因为任务过多而导致死机等情况

注意:池子中的进程或线程的数量是固定的,可以自己定义数量,但是里边的进程或线程是不会动态改变的.例如工厂中有5个工人,一堆任务过来,他们只能同时执行5个,一个工人做完一个任务之后再做另一个任务,任务在变,但是工人是不会变的,即虽然任务在变,但是进程或线程只会创建一次,执行多少任务还是那几个,这样也节省了反复开闭进程线程的时间

#1 介绍
concurrent.futures模块提供了高度封装的异步调用接口
ThreadPoolExecutor：线程池，提供异步调用
ProcessPoolExecutor: 进程池，提供异步调用
Both implement the same interface, which is defined by the abstract Executor class.

#2 基本方法
#submit(fn, *args, **kwargs)
异步提交任务

#map(func, *iterables, timeout=None, chunksize=1) 
取代for循环submit的操作

#shutdown(wait=True) 
相当于进程池的pool.close()+pool.join()操作
wait=True，等待池内所有任务执行完毕回收完资源后才继续
wait=False，立即返回，并不会等待池内的任务执行完毕
但不管wait参数为何值，整个程序都会等到所有任务执行完毕
submit和map必须在shutdown之前

#result(timeout=None)
取得结果

#add_done_callback(fn)
回调函数

# done()
判断某一个线程是否完成

# cancle()
取消某个任务

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
import time
import os

# pool = ThreadPoolExecutor(5)  # 创建线程池对象,参数可以不传,默认参数是当前计算机cpu个数*5
pool = ProcessPoolExecutor(5)  # 创建进程池对象,参数可以不传,默认参数是当前计算机cpu个数

def task(i):
    print('线程%s,进程号:%s'%(i,os.getpid()))
    time.sleep(1)  # 模拟IO
    return i

'''
提交任务的方式有两种
    同步:提交任务之后,原地等待任务的返回值,期间不做任何事
    异步:提交任务之后,不等待返回结果,继续执行下一行代码
'''

def call_back(n):
    print('回调函数拿到了返回值%s'%n.result())  # 返回值+.result()取得结果
'''
异步提交任务怎么拿到返回值:
    异步回调机制:当异步提交的任务有结果后,会自动触发回调函数的执行
'''

if __name__ == '__main__':
    t_list = []
    for i in range(20):
        t = pool.submit(task,i)
        t.add_done_callback(call_back)  # 异步回调机制
    print(os.cpu_count())  # 当前计算机是8核计算机

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor

import os,time,random
def task(n):
    print('%s is runing' %os.getpid())
    time.sleep(random.randint(1,3))
    return n**2

if __name__ == '__main__':

    executor=ThreadPoolExecutor(max_workers=3)

    # for i in range(11):
    #     future=executor.submit(task,i)

    executor.map(task,range(1,12)) #map取代了for+submit

map的用法

2.协程

　　进程是资源单位,线程是执行单位,开辟多进程和多线程都能实现并发,提高执行效率,最大化的利用计算机硬件的执行速度,但是无论是创建线程还是创建进程,都需要消耗一定的时间,随着对效率的追求,我们需要一种能够基于单线程就可以实现并发的需求,这样执行效率就能被再次提高.

　　并发的本质是:切换+保存状态

　　cpu的切换执行有两种情况,一是遇到阻塞(IO操作等),二是执行时间过长,自动切换到另一个任务,当我们使用线程时也会遇到上述两种情况,这是cpu就会切换到另一个线程执行,当前任务就会被搁置,不能最大化的利用cpu,这时候如果能在一个线程中开辟其他任务,当cpu在该线程遇到IO操作时,执行该线程中的其他任务,而不是执行其他线程,这就能最大化的提高代码的运行效率

什么是协程

　　协程就是能够在单线程下实现并发的程序

　　协程是由用户程序自己控制调度的,他不是python自带的,是我们程序员自己开发的,程序员通过代码来检测程序中的IO,一旦遇到IO自己通过代码来切换,让操作系统误认为这个程序没有任何IO,一直执行下去,如果程序中的任务没有都被执行,程序就只在运行态和就绪态之间切换,提高代码的运行效率

注意:当程序中的所有任务都处于IO阻塞状态,cpu还是会切换到其他程序运行,并不会死盯着这一个程序

协程的优点

　　1.协程的切换开销更小,是应用程序级别的,操作系统感知不到,因而更加轻量化

　　2.单线程内就能实现并发,能够最大限度的利用cpu

缺点

　　1.协程是在单线程下工作的,所以他和线程一样无法利用多核优势,我们可以多进程+多线程+多协程综合使用

　　2.协程是在但一个线程下使用的,因而一旦出现阻塞,会阻塞整个线程

特点:

　　1.可以在单线程内实现并发

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

　　3.用户程序自己切换+保存状态

　　4.一个协程遇到IO操作会自动切换到其他协程

Gevent模块

协程的使用需要引入第三方的模块Gevent

g1=gevent.spawn(func,1,,2,3,x=4,y=5)创建一个协程对象g1，spawn括号内第一个参数是函数名，如eat，后面可以有多个参数，可以是位置实参或关键字实参，都是传给函数eat的

g2=gevent.spawn(func2)

g1.join() #等待g1结束

g2.join() #等待g2结束

#或者上述两步合作一步：gevent.joinall([g1,g2])

g1.value#拿到func1的返回值

用法介绍

from gevent import monkey;monkey.patch_all()
# 需要将from gevent import monkey;monkey.patch_all()放到文件的开头
from gevent import spawn
import time

def func1():
    print('任务1开始')
    time.sleep(1)
    print('任务1结束')

def func2():
    print('任务2开始')
    time.sleep(1)
    print('任务2结束')

start = time.time()
g1 = spawn(func1)
g2 = spawn(func2)
g1.join()
g2.join()  # 执行时间1.0054426193237305
# func1()  # 函数形式正常执行
# func2()  # 执行时间2.0061123371124268
print(time.time() - start)

TCP单线程利用协程实现并发

# 单线程利用协程实现并发服务端
from gevent import monkey;monkey.patch_all()
from gevent import spawn
import socket

server = socket.socket()
server.bind(('127.0.0.1',8080))
server.listen()

def func1():
    while True:
        conn, addr = server.accept()
        spawn(func2,conn)

def func2(conn):
    while True:
        try:
            data = conn.recv(1024)
            if len(data) == 0:break
            print(data.decode('utf-8'))
            conn.send(data.upper())
        except ConnectionResetError as e:
            print(e)
            break
    conn.close()

if __name__ == '__main__':
    f1 = spawn(func1)
    f1.join()

# 单线程利用协程实现并发客户端
import socket
from threading import Thread,current_thread

client = socket.socket()
client.connect(('127.0.0.1',8080))

def task(i):
    while True:
        data = '%s线程号%s'%(i,current_thread().name)
        client.send(data.encode('utf-8'))
        res = client.recv(1024)
        print(res.decode('utf-8'))

i = 0
while True:
    t = Thread(target=task,args=(i,))
    t.start()
    i += 1

3.IO模型

1.IO模型介绍

　　为了更好地了解IO模型，我们需要事先回顾下：同步、异步、阻塞、非阻塞

    同步（synchronous） IO和异步（asynchronous） IO，阻塞（blocking） IO和非阻塞（non-blocking）IO分别是什么，到底有什么区别？这个问题其实不同的人给出的答案都可能不同，比如wiki，就认为asynchronous IO和non-blocking IO是一个东西。这其实是因为不同的人的知识背景不同，并且在讨论这个问题的时候上下文(context)也不相同。所以，为了更好的回答这个问题，我先限定一下本文的上下文。

    本文讨论的背景是Linux环境下的network IO。本文最重要的参考文献是Richard Stevens的“UNIX® Network Programming Volume 1, Third Edition: The Sockets Networking ”，6.2节“I/O Models ”，Stevens在这节中详细说明了各种IO的特点和区别，如果英文够好的话，推荐直接阅读。Stevens的文风是有名的深入浅出，所以不用担心看不懂。本文中的流程图也是截取自参考文献。

    Stevens在文章中一共比较了五种IO Model：
    * blocking IO           阻塞IO
    * nonblocking IO      非阻塞IO
    * IO multiplexing      IO多路复用
    * signal driven IO     信号驱动IO
    * asynchronous IO    异步IO
    由signal driven IO（信号驱动IO）在实际中并不常用，所以主要介绍其余四种IO Model。

    再说一下IO发生时涉及的对象和步骤。对于一个network IO (这里我们以read举例)，它会涉及到两个系统对象，一个是调用这个IO的process (or thread)，另一个就是系统内核(kernel)。当一个read操作发生时，该操作会经历两个阶段：

#1）等待数据准备 (Waiting for the data to be ready)
#2）将数据从内核拷贝到进程中(Copying the data from the kernel to the process)

　　记住这两点很重要，因为这些IO模型的区别就是在两个阶段上各有不同的情况。

2.阻塞IO(blocking IO)

　　在linux中，默认情况下所有的socket都是blocking，一个典型的读操作流程大概是这样：

　　

　　当用户进程调用了recvfrom这个系统调用，kernel就开始了IO的第一个阶段：准备数据。对于network io来说，很多时候数据在一开始还没有到达（比如，还没有收到一个完整的UDP包），这个时候kernel就要等待足够的数据到来。

    而在用户进程这边，整个进程会被阻塞。当kernel一直等到数据准备好了，它就会将数据从kernel中拷贝到用户内存，然后kernel返回结果，用户进程才解除block的状态，重新运行起来。
    所以，blocking IO的特点就是在IO执行的两个阶段（等待数据和拷贝数据两个阶段）都被block了。

    几乎所有的程序员第一次接触到的网络编程都是从listen()、send()、recv() 等接口开始的，使用这些接口可以很方便的构建服务器/客户机的模型。然而大部分的socket接口都是阻塞型的。如下图

    ps：所谓阻塞型接口是指系统调用（一般是IO接口）不返回调用结果并让当前线程一直阻塞，只有当该系统调用获得结果或者超时出错时才返回。

      

　　实际上，除非特别指定，几乎所有的IO接口 ( 包括socket接口 ) 都是阻塞型的。这给网络编程带来了一个很大的问题，如在调用recv(1024)的同时，线程将被阻塞，在此期间，线程将无法执行任何运算或响应任何的网络请求。

    一个简单的解决方案：

#在服务器端使用多线程（或多进程）。多线程（或多进程）的目的是让每个连接都拥有独立的线程（或进程），这样任何一个连接的阻塞都不会影响其他的连接。

    该方案的问题是：

#开启多进程或都线程的方式，在遇到要同时响应成百上千路的连接请求，则无论多线程还是多进程都会严重占据系统资源，降低系统对外界响应效率，而且线程与进程本身也更容易进入假死状态。

    改进方案：    

#很多程序员可能会考虑使用“线程池”或“连接池”。“线程池”旨在减少创建和销毁线程的频率，其维持一定合理数量的线程，并让空闲的线程重新承担新的执行任务。“连接池”维持连接的缓存池，尽量重用已有的连接、减少创建和关闭连接的频率。这两种技术都可以很好的降低系统开销，都被广泛应用很多大型系统，如websphere、tomcat和各种数据库等。

    改进后方案其实也存在着问题：

#“线程池”和“连接池”技术也只是在一定程度上缓解了频繁调用IO接口带来的资源占用。而且，所谓“池”始终有其上限，当请求大大超过上限时，“池”构成的系统对外界的响应并不比没有池的时候效果好多少。所以使用“池”必须考虑其面临的响应规模，并根据响应规模调整“池”的大小。

    对应上例中的所面临的可能同时出现的上千甚至上万次的客户端请求，“线程池”或“连接池”或许可以缓解部分压力，但是不能解决所有问题。总之，多线程模型可以方便高效的解决小规模的服务请求，但面对大规模的服务请求，多线程模型也会遇到瓶颈，可以用非阻塞接口来尝试解决这个问题。

 

3.非阻塞IO(non-blocking IO)

Linux下，可以通过设置socket使其变为non-blocking。当对一个non-blocking socket执行读操作时，流程是这个样子：

　　

　　从图中可以看出，当用户进程发出read操作时，如果kernel中的数据还没有准备好，那么它并不会block用户进程，而是立刻返回一个error。从用户进程角度讲 ，它发起一个read操作后，并不需要等待，而是马上就得到了一个结果。用户进程判断结果是一个error时，它就知道数据还没有准备好，于是用户就可以在本次到下次再发起read询问的时间间隔内做其他事情，或者直接再次发送read操作。一旦kernel中的数据准备好了，并且又再次收到了用户进程的system call，那么它马上就将数据拷贝到了用户内存（这一阶段仍然是阻塞的），然后返回。

    也就是说非阻塞的recvform系统调用调用之后，进程并没有被阻塞，内核马上返回给进程，如果数据还没准备好，此时会返回一个error。进程在返回之后，可以干点别的事情，然后再发起recvform系统调用。重复上面的过程，循环往复的进行recvform系统调用。这个过程通常被称之为轮询。轮询检查内核数据，直到数据准备好，再拷贝数据到进程，进行数据处理。需要注意，拷贝数据整个过程，进程仍然是属于阻塞的状态。

    所以，在非阻塞式IO中，用户进程其实是需要不断的主动询问kernel数据准备好了没有。

但是非阻塞IO模型绝不被推荐。

    我们不能否则其优点：能够在等待任务完成的时间里干其他活了（包括提交其他任务，也就是 “后台” 可以有多个任务在“”同时“”执行）。

    但是也难掩其缺点：

#1. 循环调用recv()将大幅度推高CPU占用率；这也是我们在代码中留一句time.sleep(2)的原因,否则在低配主机下极容易出现卡机情况
#2. 任务完成的响应延迟增大了，因为每过一段时间才去轮询一次read操作，而任务可能在两次轮询之间的任意时间完成。这会导致整体数据吞吐量的降低。

    此外，在这个方案中recv()更多的是起到检测“操作是否完成”的作用，实际操作系统提供了更为高效的检测“操作是否完成“作用的接口，例如select()多路复用模式，可以一次检测多个连接是否活跃。

4.多路复用IO(IO multiplexing)

 

　　IO multiplexing这个词可能有点陌生，但是如果我说select/epoll，大概就都能明白了。有些地方也称这种IO方式为事件驱动IO(event driven IO)。我们都知道，select/epoll的好处就在于单个process就可以同时处理多个网络连接的IO。它的基本原理就是select/epoll这个function会不断的轮询所负责的所有socket，当某个socket有数据到达了，就通知用户进程。它的流程如图：

　　当用户进程调用了select，那么整个进程会被block，而同时，kernel会“监视”所有select负责的socket，当任何一个socket中的数据准备好了，select就会返回。这个时候用户进程再调用read操作，将数据从kernel拷贝到用户进程。
    这个图和blocking IO的图其实并没有太大的不同，事实上还更差一些。因为这里需要使用两个系统调用(select和recvfrom)，而blocking IO只调用了一个系统调用(recvfrom)。但是，用select的优势在于它可以同时处理多个connection。

    强调：

    1. 如果处理的连接数不是很高的话，使用select/epoll的web server不一定比使用multi-threading + blocking IO的web server性能更好，可能延迟还更大。select/epoll的优势并不是对于单个连接能处理得更快，而是在于能处理更多的连接。

    2. 在多路复用模型中，对于每一个socket，一般都设置成为non-blocking，但是，如上图所示，整个用户的process其实是一直被block的。只不过process是被select这个函数block，而不是被socket IO给block。

    结论: select的优势在于可以处理多个连接，不适用于单个连接 

 select监听fd变化的过程分析：

#用户进程创建socket对象，拷贝监听的fd到内核空间，每一个fd会对应一张系统文件表，内核空间的fd响应到数据后，就会发送信号给用户进程数据已到；
#用户进程再发送系统调用，比如（accept）将内核空间的数据copy到用户空间，同时作为接受数据端内核空间的数据清除，这样重新监听时fd再有新的数据又可以响应到了（发送端因为基于TCP协议所以需要收到应答后才会清除）。

    该模型的优点：

#相比其他模型，使用select() 的事件驱动模型只用单线程（进程）执行，占用资源少，不消耗太多 CPU，同时能够为多客户端提供服务。如果试图建立一个简单的事件驱动的服务器程序，这个模型有一定的参考价值。

    该模型的缺点：

#首先select()接口并不是实现“事件驱动”的最好选择。因为当需要探测的句柄值较大时，select()接口本身需要消耗大量时间去轮询各个句柄。
#很多操作系统提供了更为高效的接口，如linux提供了epoll，BSD提供了kqueue，Solaris提供了/dev/poll，…。
#如果需要实现更高效的服务器程序，类似epoll这样的接口更被推荐。遗憾的是不同的操作系统特供的epoll接口有很大差异，
#所以使用类似于epoll的接口实现具有较好跨平台能力的服务器会比较困难。
#其次，该模型将事件探测和事件响应夹杂在一起，一旦事件响应的执行体庞大，则对整个模型是灾难性的。

 

5.异步IO(Asynchronous I/O)

Linux下的asynchronous IO其实用得不多，从内核2.6版本才开始引入。先看一下它的流程：

　　用户进程发起read操作之后，立刻就可以开始去做其它的事。而另一方面，从kernel的角度，当它受到一个asynchronous read之后，首先它会立刻返回，所以不会对用户进程产生任何block。然后，kernel会等待数据准备完成，然后将数据拷贝到用户内存，当这一切都完成之后，kernel会给用户进程发送一个signal，告诉它read操作完成了。

6.IO模型比较分析

到目前为止，已经将四个IO Model都介绍完了。现在回过头来回答最初的那几个问题：blocking和non-blocking的区别在哪，synchronous IO和asynchronous IO的区别在哪。
    先回答最简单的这个：blocking vs non-blocking。前面的介绍中其实已经很明确的说明了这两者的区别。调用blocking IO会一直block住对应的进程直到操作完成，而non-blocking IO在kernel还准备数据的情况下会立刻返回。

    再说明synchronous IO和asynchronous IO的区别之前，需要先给出两者的定义。Stevens给出的定义（其实是POSIX的定义）是这样子的：
    A synchronous I/O operation causes the requesting process to be blocked until that I/O operationcompletes;
    An asynchronous I/O operation does not cause the requesting process to be blocked; 
    两者的区别就在于synchronous IO做”IO operation”的时候会将process阻塞。按照这个定义，四个IO模型可以分为两大类，之前所述的blocking IO，non-blocking IO，IO multiplexing都属于synchronous IO这一类，而 asynchronous I/O后一类 。

    有人可能会说，non-blocking IO并没有被block啊。这里有个非常“狡猾”的地方，定义中所指的”IO operation”是指真实的IO操作，就是例子中的recvfrom这个system call。non-blocking IO在执行recvfrom这个system call的时候，如果kernel的数据没有准备好，这时候不会block进程。但是，当kernel中数据准备好的时候，recvfrom会将数据从kernel拷贝到用户内存中，这个时候进程是被block了，在这段时间内，进程是被block的。而asynchronous IO则不一样，当进程发起IO 操作之后，就直接返回再也不理睬了，直到kernel发送一个信号，告诉进程说IO完成。在这整个过程中，进程完全没有被block。

    各个IO Model的比较如图所示：

　　

　　经过上面的介绍，会发现non-blocking IO和asynchronous IO的区别还是很明显的。在non-blocking IO中，虽然进程大部分时间都不会被block，但是它仍然要求进程去主动的check，并且当数据准备完成以后，也需要进程主动的再次调用recvfrom来将数据拷贝到用户内存。而asynchronous IO则完全不同。它就像是用户进程将整个IO操作交给了他人（kernel）完成，然后他人做完后发信号通知。在此期间，用户进程不需要去检查IO操作的状态，也不需要主动的去拷贝数据。

 33