python基础——并发编程

进程
以一个整体的形式暴露给操作系统管理，里面包含对各种资源的调用，内存的对各种资源管理的集合，就可以称为进程，进程要操作CPU，必须要先创建一个线程。一个进程可以并发多个线程，每条线程执行不同的任务。启动一个进程的时候，会自动启动一个线程。进程间的内存是独立的。父进程创建子进程，子进程只是克隆一份父进程。

线程
线程是操作系统最小的调度单位，是一串指令的集合，所有在同一个进程里的线程是共享同一块内存空间的。进程里的第一个线程就是主线程。启动线程比启动进程快。

进程与线程的区别
进程间的内存是独立的。父进程创建子进程，子进程只是克隆一份父进程。一个进程里的线程是共享同一块内存空间的。启动线程比启动进程快。同一个进程的线程之间可以直接交流，两个进程想通信，必须通过一个中间代理来实现。创建新线程很简单，创建新进程需要对其父进程进行一次克隆，一个线程可以控制和操作同一进程里的其他线程，但是进程只能操作子进程。

创建线程

import threading
import time

def run(n):
    print("task",n)
    time.sleep(2)
t1=threading.Thread(target=run,args=("t1",))#创建线程1
t2=threading.Thread(target=run,args=("t1",))#创建线程2
#并发执行一共只等2秒
t1.start()
t2.start()

用类实现的线程创建

import threading
import time
class MyThread(threading.Thread):
    def __init__(self,n):
        super(MyThread,self).__init__()
        self.n=n
    def run(self):
        print("running task",self.n)
t1=MyThread("t1")
t2=MyThread("t2")
t1.start()
t2.start()

有关join的案例

import threading
import time

def run(n):
    print("task",n)
    time.sleep(2)

start_time=time.time()
for i in range(50):#创建线程50个线程
    t=threading.Thread(target=run,args=("t-%s"%i,))
    t.start()

print("cost:",time.time()-start_time)#cost: 0.00500035285949707，这个时间并不是2秒，因为这个程序本身就是一个主线程，和其他的线程是独立的，它启动完子线程之后继续运行，它们是独立的，并行的，而不会像串行那样等着子线程运行完再执行后面的，所以这样是测不出来的

import threading
import time

def run(n):
    print("task",n)
    time.sleep(2)

res=[]#存线程实例
start_time=time.time()
for i in range(50):#创建线程50个线程
    t=threading.Thread(target=run,args=("t-%s"%i,))
    t.start()
    res.append(t)

for r in res:#循环线程实例列表，等待所有的线程执行完毕
    r.join()#线程执行完毕后，才会往后执行，相当于C语言中的wait()
print("--------all threads has finished...")
print("cost:",time.time()-start_time)#结果是  cost: 2.006114959716797

守护线程

守护线程是为主线程服务的，主线程并不管子线程有没有运行结束，当主线程结束之后，程序结束了。

import threading
import time

def run(n):
    print("task",n)
    time.sleep(2)

res=[]#存线程实例
start_time=time.time()
for i in range(50):#创建线程50个线程
    t=threading.Thread(target=run,args=("t-%s"%i,))
    t.setDaemon(True)#将当前线程变成主线程的守护线程，这些守护线程是为主线程服务的，主线程运行完毕了程序就结束了，它不会等待子线程运行完毕再结束退出（可以看到结果，程序在退出之前没有停顿2秒），只能在start之前设置
    t.start()
    res.append(t)

# for r in res:#循环线程实例列表，等待所有的线程执行完毕
#     r.join()#线程执行完毕后，才会往后执行，相当于C语言中的wait()
print("--------all threads has finished...")
print("cost:",time.time()-start_time)#结果是  cost: 2.006114959716797

锁

import threading
import time

def run(n):
    lock.acquire()#获取锁
    global  num
    # time.sleep(0.8)#sleep的时候不占用CPU，只是把它挂起了，等一段时间检查下
    num=num+1
    lock.release()#释放锁，锁没释放的时候其他线程就不能使用

lock=threading.Lock()#生成锁
num=0
res=[]#存线程实例
start_time=time.time()
for i in range(50):#创建线程50个线程
    t=threading.Thread(target=run,args=("t-%s"%i,))
    t.start()
    res.append(t)

for r in res:#循环线程实例列表，等待所有的线程执行完毕
    r.join()#线程执行完毕后，才会往后执行，相当于C语言中的wait()
print("--------all threads has finished...")
print("num:",num)

递归锁

import threading,time

def run1():
    print("grab the first part data")
    lock.acquire()
    global num
    num +=1
    lock.release()
    return num
def run2():
    print("grab the second part data")
    lock.acquire()
    global  num2
    num2+=1
    lock.release()
    return num2
def run3():
    lock.acquire()
    res = run1()
    print('--------between run1 and run2-----')
    res2 = run2()
    lock.release()
    print(res,res2)


if __name__ == '__main__':

    num,num2 = 0,0
    #lock = threading.Lock()#这种锁直接锁死，因为大锁外面和里面的小锁解锁混了
    lock = threading.RLock()#鉴于这种锁里又套锁的使用递归锁解决
    for i in range(1):
        t = threading.Thread(target=run3)
        t.start()

while threading.active_count() != 1:#线程只有一个的时候，说明其他线程都执行完了，这个剩下的其实就是主线程
    print(threading.active_count())
else:
    print('----all threads done---')
    print(num,num2)

信号量

import threading,time

def run(n):
    semaphore.acquire()#获取信号量
    time.sleep(1)
    print("run the thread: %s\n" %n)
    semaphore.release()#释放信号量

if __name__ == '__main__':

    # num= 0
    semaphore  = threading.BoundedSemaphore(5) #最多允许5个线程同时运行
    for i in range(20):
        t = threading.Thread(target=run,args=(i,))
        t.start()

while threading.active_count() != 1:
    pass #print threading.active_count()
else:
    print('----all threads done---')
    # print(num)
'''
我们看到的效果是5个线程5个线程一拨的执行，但实际情况并非这样的，它只是一次只能运行5个线程而已，
假如5个线程中有3个线程的处理的计算少一点，先执行完了，那么又会接着运行3个线程，而并不是等着其他两个
线程都运行完了再启用5个新的进程

所谓的8核、多核意思就是同一时间最多能运行8个或多个进程，而单核只能运行一个进程，但我们肉眼看到的幻觉是
多个进程在同时进行，这是因为计算的CPU切换速度非常快，这个进程运行一会儿挂起，又运行另一个进程，来来回回
切换，速度非常快，所以我们感觉是多进程在同时运行，其实不然。

Cpython有一个缺陷就是，它并不能多核运算，一次只能运行一个线程

run the thread: 1

run the thread: 2

run the thread: 3
run the thread: 0
run the thread: 4
'''

事件（红绿灯）

import time
import threading
event=threading.Event()
def lighter():
    count=0
    event.set()#先设为绿灯
    while True:
        if count>5 and count<10:#改成红灯
            event.clear()#把标志位清空，就会等待
            print("\033[41;1mred light is on....\033[0m")
        elif count>10:
            event.set()#变绿灯
            count=0
        else:
            print("\033[42;1mgreen light is on....\033[0m")
        time.sleep(1)
        count+=1
def car(name):
    while True:
        if event.is_set():#是否有设置标志位，有，则代表绿灯
            print("[%s] running..."%name)
            time.sleep(1)
        else:
            print("[%s] sees red light,waiting..."%name)
            event.wait()#等待标志位被设置（等待绿灯），阻塞
            print("\033[34;1m[%s] green light is on,start going...\033[0m"%name)

light=threading.Thread(target=lighter,)
light.start()

car1=threading.Thread(target=car,args=("Tesla",))#有参数就传递参数args=("Tesla",)
car1.start()

队列

队列的作用：解耦，使程序直接实现松耦合，提高处理效率

import threading,time
import queue

q=queue.Queue(maxsize=10)#设置对列最大长度为10
def Producer(name):
    count=1
    while True:
        for i in range(10):
            q.put("骨头%s"%i)#往队列里面放数据
            print("生产了%s骨头"%count)
            count+=1

def Consumer(name):
    while True:
        print("[%s]取到[%s]并且吃了它……"%(name,q.get()))
        time.sleep(1)

p=threading.Thread(target=Producer,args=("Alex",))
c=threading.Thread(target=Consumer,args=("dog",))
p.start()
c.start()
#q.get_nowait()#当对列里没有数据的时候用q.get()会阻塞，但用这个不会
#q.put_nowait()#跟上面一样，当队列满了的时候，用q.put()会阻塞，但用这个不会
#q.LifoQueue #后进先出队列
#q.PriorityQueue()#存储数据时可设置优先级的队列

io操作不占用CPU，计算占用CPU，python多线程不适合CPU密集操作型的任务，适合IO操作密集型的任务

进程

from multiprocessing import Process
import os

def info(title):
    print(title)
    print('module name:', __name__)
    print('parent process:', os.getppid())#获取父进程ID（这里是pycharm的进程号，改该进程是由父进程pycharm生成的）
    print('process id:', os.getpid())#获取子进程ID（也就是主程序的进程号，因为子进程都是由父进程生成的）
    print("\n\n")

def f(name):
    info('\033[31;1mfunction f\033[0m')
    print('hello', name)

if __name__ == '__main__':
    info('\033[32;1mmain process line\033[0m')
    p = Process(target=f, args=('bob',))#启动一个子进程f
    p.start()
    p.join()

多进程

import multiprocessing
import threading
import time
#进程里面调用线程
def run(name):
    time.sleep(2)
    print("hello",name)
    t=threading.Thread(target=thread_run,)#进程里面调用线程
    t.start()
def thread_run():
    print(threading.get_ident())
if __name__=="__main__":
    for i in range(10):#启动10个进程
        p=multiprocessing.Process(target=run,args=("job %s" %i,))#启动一个进程
        p.start()
        # p.join()

进程间的通信

进程队列

#进程queue，直接导入的Queue是线程Queue
from multiprocessing import Process, Queue

def f(qq):
    qq.put([42, None, 'hello'])
#对于进程来说，父、子进程间的内存是相互独立的，内存不共享，也就是不能像线程queue间那样数据共享
#而进程queue则可以，它的实现其实就是先把子进程的queue克隆一份放在内存的中，通过序列化实现的，当父进程
#要用到子进程的queue时将之前序列化在内存中的数据进行反序列化出来实现的

if __name__ == '__main__':
    q = Queue()
    p = Process(target=f, args=(q,))#子进程的q
    p.start()
    print(q.get())    # prints "[42, None, 'hello']"，父进程能取到子进程的数据
    p.join()

进程间的数据修改（进程字典、列表）

from multiprocessing import Process, Manager
import os

def f(d, l):
    d[os.getpid()]=os.getpid()#将每个进程的ID放入字典中
    l.append(os.getpid())#将每个进程的ID放入列表中
    print(l)

if __name__ == '__main__':
    with Manager() as manager:
        d = manager.dict()#生成可在多个进程间传递共享的字典

        l = manager.list(range(5))#生成可在多个进程间传递共享的列表
        p_list = []
        for i in range(10):
            p = Process(target=f, args=(d, l))
            p.start()
            p_list.append(p)
        for res in p_list:
            res.join()

        print(d)
        print(l)

进程池

from  multiprocessing import Process,Pool
import os
import time


def Foo(i):
    time.sleep(2)
    print("in process",os.getpid())
    return i+100

def Bar(arg):
    print('-->exec done:',arg,os.getpid())
if __name__=="__main__":
    pool = Pool(3)#允许进程池里同时放入5个进程
    print("主进程的PID：",os.getpid())
    for i in range(10):
        pool.apply_async(func=Foo, args=(i,),callback=Bar)#异步执行，即并行，callback是回调，func干完之后再干callback，没干完func就不干callback（是由父进程进行回调的）
        # pool.apply(func=Foo, args=(i,))#同步执行，即串行

    print('end')
    pool.close()
    pool.join()#进程池中进程执行完毕后再关闭（等待所有进程执行完毕），如果注释，那么程序直接关闭。

管道

from multiprocessing import Process, Pipe

def f(conn):
    conn.send([42, None, 'hello'])#子进程向父进程发送一条数据
    conn.send([42, None, 'hello'])#子进程向父进程发送一条数据，如果子进程发送两条，而父进程只接收一条则父进程会等待
    print("这是来自父进程的数据:",conn.recv())#子进程接收来自父进程的数据
    conn.close()
#管道有点像socket,电话线一样，一边发，一边接收
if __name__ == '__main__':
    parent_conn, child_conn = Pipe()#生成管道实例，有两个返回值
    p = Process(target=f, args=(child_conn,))
    p.start()
    print(parent_conn.recv())   # prints "[42, None, 'hello']" 父进程接收子进程发送过来的数据
    print("父进程接收第二条数据%s"%parent_conn.recv())
    parent_conn.send("hello my children")#父进程向子进程发送数据
    p.join()

进程锁
防止多个进程对同一个对象进行操作的时候出现紊乱。

from multiprocessing import Process, Lock
#这个锁存在的意义就是，在打印数据的时候防止争抢着打印数据，导致某个进程还没打印完，另一个进程就开始打印（在Linux上能试出来）
def f(l, i):
    l.acquire()
    try:
        print('hello world', i)
    finally:
        l.release()

if __name__ == '__main__':
    lock = Lock()

    for num in range(10):
        Process(target=f, args=(lock, num)).start()

协程

协程，又称微线程，纤程。英文名Coroutine。一句话说明什么是线程：协程是一种用户态的轻量级线程。

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

协程能保留上一次调用时的状态（即所有局部状态的一个特定组合），每次过程重入时，就相当于进入上一次调用的状态，换种说法：进入上一次离开时所处逻辑流的位置。遇到IO操作就切换！

协程的好处：

无需线程上下文切换的开销
无需原子操作锁定及同步的开销
“原子操作(atomic operation)是不需要synchronized”，所谓原子操作是指不会被线程调度机制打断的操作；这种操作一旦开始，就一直运行到结束，中间不会有任何 context switch （切换到另一个线程）。原子操作可以是一个步骤，也可以是多个操作步骤，但是其顺序是不可以被打乱，或者切割掉只执行部分。视作整体是原子性的核心。
方便切换控制流，简化编程模型
高并发+高扩展性+低成本：一个CPU支持上万的协程都不是问题。所以很适合用于高并发处理。

缺点：

无法利用多核资源：协程的本质是个单线程,它不能同时将 单个CPU 的多个核用上,协程需要和进程配合才能运行在多CPU上.当然我们日常所编写的绝大部分应用都没有这个必要，除非是cpu密集型应用。
进行阻塞（Blocking）操作（如IO时）会阻塞掉整个程序

from gevent import greenlet
#协程可以实现自动切换，该程序先调用test1,打印12,然后切换到gr2打印56，再切换回来到gr1打印34,又切换回去gr2打印78
def test1():
    print(12)
    gr2.switch()
    print(34)
    gr2.switch()


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


gr1 = greenlet.greenlet(test1)
gr2 = greenlet.greenlet(test2)
gr1.switch()

import gevent
import time

#程序执行顺序：先执行位置A，然后遇到位置B有IO操作，切换到位置D，然后遇到E位置的IO操作切换到位置B，位置B有IO切换就又切换到
#位置E，它们之间来回切换，直到位置E的1秒结束之后，执行位置F，执行完之后又切换到位置B，此时func2没有要执行的了，就在此等待1秒
#然后执行位置C（gevent.sleep（）一直在计时，并非没有执行）
#总体执行顺序如下：A D F C

def func1():
    print('\033[31;1m李闯在跟海涛搞...\033[0m')#位置A
    start_time=time.time()
    gevent.sleep(2)#位置B
    end_time=time.time()
    print("func1总共花了%s时间"%(end_time-start_time))
    print('\033[31;1m李闯又回去跟继续跟海涛搞...\033[0m')#位置C

def func2():
    print('\033[32;1m李闯切换到了跟海龙搞...\033[0m')#位置D
    start_time=time.time()
    gevent.sleep(1)#位置E
    end_time=time.time()
    print("func2总共花了%s时间"%(end_time-start_time))
    print('\033[32;1m李闯搞完了海涛，回来继续跟海龙搞...\033[0m')#位置F



gevent.joinall([
    gevent.spawn(func1),#启动一个协程
    gevent.spawn(func2),
    #gevent.spawn(func3),
])

简单爬虫

from gevent import monkey; monkey.patch_all()#把当前程序的所有IO操作给我单独的做上标记
#如果没有给gevent打上补丁的话，它是检测不到除gevent它本省自带的IO操作的，当打上了补丁，它就能检测到程序其他所有的IO操作
import gevent
import time
from  urllib.request import urlopen

def f(url):
    print('GET: %s' % url)
    resp = urlopen(url)
    data = resp.read()#下载下来的网页
    # f=open("URL.html","wb")
    # f.write(data)
    # f.close()
    print('%d bytes received from %s.' % (len(data), url))
start_time=time.time()
gevent.joinall([
        gevent.spawn(f, 'https://www.python.org/'),#三个都切换f,后面的参数是给函数f传递的参数
        gevent.spawn(f, 'https://www.yahoo.com/'),
        gevent.spawn(f, 'https://github.com/'),
])
end_time=time.time()
print("总共花了多长时间：",(end_time-start_time))

#没有打补丁的程序就是串行的效果，结果如下图：
'''
GET: https://www.python.org/
48769 bytes received from https://www.python.org/.
GET: https://www.yahoo.com/
498102 bytes received from https://www.yahoo.com/.
GET: https://github.com/
55997 bytes received from https://github.com/.
总共花了多长时间： 5.136293888092041
'''
#打了补丁的程序就是异步并行的效果，结果如图：
'''
GET: https://www.python.org/
GET: https://www.yahoo.com/
GET: https://github.com/
493940 bytes received from https://www.yahoo.com/.
48769 bytes received from https://www.python.org/.
55989 bytes received from https://github.com/.
总共花了多长时间： 3.253185987472534
'''

事件驱动与异步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操作上所花费的时间已经用灰色框标示出来了。

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

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

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

当我们面对如下的环境时，事件驱动模型通常是一个好的选择：
1、程序中有许多任务，而且…
2、任务之间高度独立（因此它们不需要互相通信，或者等待彼此）而且…
3、在等待事件到来时，某些任务会阻塞。

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

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

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

概念说明

用户空间与内核空间

现在操作系统都是采用虚拟存储器，那么对32位操作系统而言，它的寻址空间（虚拟存储空间）为4G（2的32次方）。操作系统的核心是内核，独立于普通的应用程序，可以访问受保护的内存空间，也有访问底层硬件设备的所有权限。为了保证用户进程不能直接操作内核（kernel），保证内核的安全，操心系统将虚拟空间划分为两部分，一部分为内核空间，一部分为用户空间。针对linux操作系统而言，将最高的1G字节（从虚拟地址0xC0000000到0xFFFFFFFF），供内核使用，称为内核空间，而将较低的3G字节（从虚拟地址0x00000000到0xBFFFFFFF），供各个进程使用，称为用户空间。

进程切换

为了控制进程的执行，内核必须有能力挂起正在CPU上运行的进程，并恢复以前挂起的某个进程的执行。这种行为被称为进程切换。因此可以说，任何进程都是在操作系统内核的支持下运行的，是与内核紧密相关的。

从一个进程的运行转到另一个进程上运行，这个过程中经过下面这些变化：
1. 保存处理机上下文，包括程序计数器和其他寄存器。
2. 更新PCB信息。
3. 把进程的PCB移入相应的队列，如就绪、在某事件阻塞等队列。
4. 选择另一个进程执行，并更新其PCB。
5. 更新内存管理的数据结构。
6. 恢复处理机上下文。
总而言之就是很耗资源

进程的阻塞

正在执行的进程，由于期待的某些事件未发生，如请求系统资源失败、等待某种操作的完成、新数据尚未到达或无新工作做等，则由系统自动执行阻塞原语(Block)，使自己由运行状态变为阻塞状态。可见，进程的阻塞是进程自身的一种主动行为，也因此只有处于运行态的进程（获得CPU），才可能将其转为阻塞状态。当进程进入阻塞状态，是不占用CPU资源的。

文件描述符fd

文件描述符（File descriptor）是计算机科学中的一个术语，是一个用于表述指向文件的引用的抽象化概念。

文件描述符在形式上是一个非负整数。实际上，它是一个索引值，指向内核为每一个进程所维护的该进程打开文件的记录表。当程序打开一个现有文件或者创建一个新文件时，内核向进程返回一个文件描述符。在程序设计中，一些涉及底层的程序编写往往会围绕着文件描述符展开。但是文件描述符这一概念往往只适用于UNIX、Linux这样的操作系统。

缓存 I/O

缓存 I/O 又被称作标准 I/O，大多数文件系统的默认 I/O 操作都是缓存 I/O。在 Linux 的缓存 I/O 机制中，操作系统会将 I/O 的数据缓存在文件系统的页缓存（ page cache ）中，也就是说，数据会先被拷贝到操作系统内核的缓冲区中，然后才会从操作系统内核的缓冲区拷贝到应用程序的地址空间。

缓存 I/O 的缺点：
数据在传输过程中需要在应用程序地址空间和内核进行多次数据拷贝操作，这些数据拷贝操作所带来的 CPU 以及内存开销是非常大的。

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。

阻塞 I/O（blocking IO）

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

当用户进程调用了recvfrom这个系统调用，kernel就开始了IO的第一个阶段：准备数据（对于网络IO来说，很多时候数据在一开始还没有到达。比如，还没有收到一个完整的UDP包。这个时候kernel就要等待足够的数据到来）。这个过程需要等待，也就是说数据被拷贝到操作系统内核的缓冲区中是需要一个过程的。而在用户进程这边，整个进程会被阻塞（当然，是进程自己选择的阻塞）。当kernel一直等到数据准备好了，它就会将数据从kernel中拷贝到用户内存，然后kernel返回结果，用户进程才解除block的状态，重新运行起来。简而言之，就是建立了连接如果没有数据返回就一直等待（阻塞），内核等到所有数据准备好才拷贝到用户内存（阻塞）。

所以，blocking IO的特点就是在IO执行的两个阶段都被block了。

非阻塞 I/O（nonblocking IO）

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

当用户进程发出read操作时，如果kernel中的数据还没有准备好，那么它并不会block用户进程，而是立刻返回一个error。从用户进程角度讲 ，它发起一个read操作后，并不需要等待，而是马上就得到了一个结果。用户进程判断结果是一个error时，它就知道数据还没有准备好，于是它可以再次发送read操作。一旦kernel中的数据准备好了，并且又再次收到了用户进程的system call，那么它马上就将数据拷贝到了用户内存，然后返回。简而言之，就是建立了连接如果没有数据返回就一直询问，内核等到所有数据准备好才拷贝到用户内存（阻塞）。

所以，nonblocking IO的特点是用户进程需要不断的主动询问kernel数据好了没有。

I/O 多路复用（ IO multiplexing）

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

当用户进程调用了select，那么整个进程会被block，而同时，kernel会“监视”所有select负责的socket，当任何一个socket中的数据准备好了，select就会返回。这个时候用户进程再调用read操作，将数据从kernel拷贝到用户进程。

所以，I/O 多路复用的特点是通过一种机制一个进程能同时等待多个文件描述符，而这些文件描述符（套接字描述符）其中的任意一个进入读就绪状态，select()函数就可以返回。

这个图和blocking IO的图其实并没有太大的不同，事实上，还更差一些。因为这里需要使用两个system call (select 和 recvfrom)，而blocking IO只调用了一个system call (recvfrom)。但是，用select的优势在于它可以同时处理多个connection。

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

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

异步 I/O（asynchronous IO）

inux下的asynchronous IO其实用得很少。先看一下它的流程：

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

总结

blocking和non-blocking的区别

调用blocking IO会一直block住对应的进程直到操作完成，而non-blocking IO在kernel还准备数据的情况下会立刻返回。

synchronous IO和asynchronous IO的区别

在说明synchronous IO和asynchronous IO的区别之前，需要先给出两者的定义。POSIX的定义是这样子的：
- A synchronous I/O operation causes the requesting process to be blocked until that I/O operation completes;
- An asynchronous I/O operation does not cause the requesting process to be blocked;

两者的区别就在于synchronous IO做”IO operation”的时候会将process阻塞。按照这个定义，之前所述的blocking IO，non-blocking IO，IO multiplexing都属于synchronous IO。

有人会说，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操作的状态，也不需要主动的去拷贝数据。

I/O 多路复用之select、poll、epoll详解

select，poll，epoll都是IO多路复用的机制。I/O多路复用就是通过一种机制，一个进程可以监视多个描述符，一旦某个描述符就绪（一般是读就绪或者写就绪），能够通知程序进行相应的读写操作。但select，poll，epoll本质上都是同步I/O，因为他们都需要在读写事件就绪后自己负责进行读写，也就是说这个读写过程是阻塞的，而异步I/O则无需自己负责进行读写，异步I/O的实现会负责把数据从内核拷贝到用户空间。

select

select(rlist, wlist, xlist, timeout=None)

select 函数监视的文件描述符分3类，分别是writefds、readfds、和exceptfds。调用后select函数会阻塞，直到有描述副就绪（有数据 可读、可写、或者有except），或者超时（timeout指定等待时间，如果立即返回设为null即可），函数返回。当select函数返回后，可以 通过遍历fdset，来找到就绪的描述符。

select目前几乎在所有的平台上支持，其良好跨平台支持也是它的一个优点。select的一 个缺点在于单个进程能够监视的文件描述符的数量存在最大限制，在Linux上一般为1024，可以通过修改宏定义甚至重新编译内核的方式提升这一限制，但 是这样也会造成效率的降低。

epoll

epoll是在2.6内核中提出的，是之前的select和poll的增强版本。相对于select和poll来说，epoll更加灵活，没有描述符限制。epoll使用一个文件描述符管理多个描述符，将用户关系的文件描述符的事件存放到内核的一个事件表中，这样在用户空间和内核空间的copy只需一次。

epoll操作过程

epoll操作过程需要三个接口，分别如下：

int epoll_create(int size)；//创建一个epoll的句柄，size用来告诉内核这个监听的数目一共有多大
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event)；
int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);

1. int epoll_create(int size);

创建一个epoll的句柄，size用来告诉内核这个监听的数目一共有多大，这个参数不同于select()中的第一个参数，给出最大监听的fd+1的值，参数size并不是限制了epoll所能监听的描述符最大个数，只是对内核初始分配内部数据结构的一个建议。
当创建好epoll句柄后，它就会占用一个fd值，在linux下如果查看/proc/进程id/fd/，是能够看到这个fd的，所以在使用完epoll后，必须调用close()关闭，否则可能导致fd被耗尽。

2. int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event)；

函数是对指定描述符fd执行op操作。
- epfd：是epoll_create()的返回值。
- op：表示op操作，用三个宏来表示：添加EPOLL_CTL_ADD，删除EPOLL_CTL_DEL，修改EPOLL_CTL_MOD。分别添加、删除和修改对fd的监听事件。
- fd：是需要监听的fd（文件描述符）
- epoll_event：是告诉内核需要监听什么事

3. int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);

等待epfd上的io事件，最多返回maxevents个事件。
参数events用来从内核得到事件的集合，maxevents告之内核这个events有多大，这个maxevents的值不能大于创建epoll_create()时的size，参数timeout是超时时间（毫秒，0会立即返回，-1将不确定，也有说法说是永久阻塞）。该函数返回需要处理的事件数目，如返回0表示已超时。

socket协程

服务端

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)



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)

客户端

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()

select实现的多路并发（同步）

服务端

import select
import socket
import queue

server=socket.socket()
server.bind(("localhost",9000))
server.listen(1000)

server.setblocking(False)#设置为非阻塞模式，没有连接就报错
inputs=[server,]#监测自己
#inputs=[server,coon]#因为select底层不知道返回的是哪个连接，所以需要将所有的连接放在这里头，让select.select去监测新来的连接是哪个
outputs=[]
while True:
    #有连接返回3个数据，新来的连接出现在readable，异常连接出现在execeptional
    readable,writeable,execeptional=select.select(inputs,outputs,inputs)#前面的inputs是指所有监测的连接，有一个活得就不阻塞，后面这个inputs是指所有监测的连接中可能有异常的
    print(readable,writeable,execeptional)
    for r in readable:
        if r is server:#代表来了一个新连接
            conn,addr=server.accept()
            print("来了个新连接",addr)
            # print(conn,addr)
            inputs.append(conn)#是因为这个新建立的连接还没发数据过来，现在就接收的话程序就报错了
            #所以要想实现这个客户端发数据来时server端能知道，就需要让select再监测这个连接conn
        else:
            data=r.recv(1024)
            print("收到数据",data)
            r.send(data)
            print("send done……")

客户端

import socket

HOST = 'localhost'    # The remote host
PORT = 9000         # 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()

加了队列的服务端

import select
import socket
import queue

server=socket.socket()
server.bind(("localhost",9000))
server.listen(1000)

server.setblocking(False)#设置为非阻塞模式，没有连接就报错
inputs=[server,]#监测自己
#inputs=[server,coon]#因为select底层不知道返回的是哪个连接，所以需要将所有的连接放在这里头，让select.select去监测新来的连接是哪个
outputs=[]
msg_dic={}
while True:
    #有连接返回3个数据，新来的连接出现在readable，异常连接出现在execeptional，要返回给客户端的连接列表writeable
    readable,writeable,execeptional=select.select(inputs,outputs,inputs)#前面的inputs是指所有监测的连接，有一个活得就不阻塞，后面这个inputs是指所有监测的连接中可能有异常的
    print(readable,writeable,execeptional)
    for r in readable:
        if r is server:#代表来了一个新连接
            conn,addr=server.accept()
            print("来了个新连接",addr)
            # print(conn,addr)
            inputs.append(conn)#是因为这个新建立的连接还没发数据过来，现在就接收的话程序就报错了
            #所以要想实现这个客户端发数据来时server端能知道，就需要让select再监测这个连接conn
            msg_dic[conn]=queue.Queue()#初始化一个队列，后面存要返回给客户端的数据
        else:
            data=r.recv(1024)
            print("收到数据",data)
            msg_dic[r].put(data)
            outputs.append(r)#放入返回的连接队列

    for w in writeable:#要返回给客户端的连接列表
        data_to_client=msg_dic[w].get()
        w.send(data_to_client)#返回给客户端源数据
        outputs.remove(w)#确保下次循环的时候writeable不返回这个已经处理完的连接

    for e in execeptional:#连接断开
        for e in outputs:
            outputs.remove(e)
        inputs.remove(e)
        del msg_dic[e]

selectors模块实现的服务端

import selectors
import socket
#默认用epoll，如果系统找不到epoll就用select
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)#新连接注册read回调函数

def read(conn, mask):
    data = conn.recv(1024)  # 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)#将连接注册到进去，只要来了新连接就调用accept

while True:
    events = sel.select()#默认是阻塞，有活得连接就返回活动的连接列表
    for key, mask in events:
        callback = key.data#相当于调用accept
        callback(key.fileobj, mask)#key.fileobj=文件句柄  执行