python并发编程之协程
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
1、协程: 单线程实现并发 在应用程序里控制多个任务的切换+保存状态 优点: 应用程序级别速度要远远高于操作系统的切换 缺点: 多个任务一旦有一个阻塞没有切,整个线程都阻塞在原地 该线程内的其他的任务都不能执行了 一旦引入协程,就需要检测单线程下所有的IO行为, 实现遇到IO就切换,少一个都不行,以为一旦一个任务阻塞了,整个线程就阻塞了, 其他的任务即便是可以计算,但是也无法运行了 2、协程的目的: 想要在单线程下实现并发 可利用的cpu只有一个 并发本质=切换+保存状态 并发指的是多个任务看起来是同时运行的
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
from gevent import monkey,spawn;monkey.patch_all() from threading import current_thread import time def eat(): print('%s eat 1' %current_thread().name) time.sleep(3) print('%s eat 2' %current_thread().name) def play(): print('%s play 1' %current_thread().name) time.sleep(1) print('%s play 2' %current_thread().name) g1=spawn(eat,) g2=spawn(play,) print(current_thread().name) g1.join() g2.join() 并发套接字通信 server: from gevent import spawn,monkey;monkey.patch_all() from socket import * from threading import Thread def talk(conn): while True: try: data=conn.recv(1024) if len(data) == 0:break conn.send(data.upper()) except ConnectionResetError: break conn.close() def server(ip,port,backlog=5): server = socket(AF_INET, SOCK_STREAM) server.bind((ip, port)) server.listen(backlog) print('starting...') while True: conn, addr = server.accept() spawn(talk, conn,) if __name__ == '__main__': g=spawn(server,'127.0.0.1',8080) g.join() client: from threading import Thread,current_thread from socket import * import os def task(): client=socket(AF_INET,SOCK_STREAM) client.connect(('127.0.0.1',8080)) while True: msg='%s say hello' %current_thread().name client.send(msg.encode('utf-8')) data=client.recv(1024) print(data.decode('utf-8')) if __name__ == '__main__': for i in range(500): t=Thread(target=task) t.start()
python并发编程之IO模型
1 阻塞IO(blocking IO)
2 非阻塞IO(non-blocking IO)
3 多路复用IO(IO multiplexing)
4 异步IO(Asynchronous I/O)
1 阻塞IO(blocking IO)
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
''' 网络IO: recvfrom: wait data:等待客户端产生数据——》客户端OS--》网络--》服务端操作系统缓存 copy data:由本地操作系统缓存中的数据拷贝到应用程序的内存中 send: copy data ''' # conn.recv(1024) ==>OS 在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)的同时,线程将被阻塞,在此期间,线程将无法执行任何运算或响应任何的网络请求。
2 非阻塞IO(non-blocking IO) 代码演示
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
Linux下,可以通过设置socket使其变为non-blocking 从图中可以看出,当用户进程发出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()多路复用模式,可以一次检测多个连接是否活跃。
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
from socket import * import time server = socket(AF_INET, SOCK_STREAM) server.bind(('127.0.0.1',8080)) server.listen(5) server.setblocking(False) conn_l=[] while True: try: print('总连接数[%s]' % len(conn_l)) conn,addr=server.accept() conn_l.append(conn) except BlockingIOError: del_l=[] for conn in conn_l: try: data=conn.recv(1024) if len(data) == 0: del_l.append(conn) continue conn.send(data.upper()) except BlockingIOError: pass except ConnectionResetError: del_l.append(conn) for conn in del_l: conn_l.remove(conn)
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
from socket import * import os client=socket(AF_INET,SOCK_STREAM) client.connect(('127.0.0.1',8080)) while True: msg='%s say hello' %os.getpid() client.send(msg.encode('utf-8')) data=client.recv(1024) print(data.decode('utf-8'))
上节课复习
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
上节课复习: 1、GIL全局解释器锁 什么是? GIL本质就是一个把互斥锁,是将多个并发的线程对共享数据的修改 变成“串行” 为何有? Cpython解释器的垃圾回收机制不是线程安全的 如何用? 有GIL的存在,导致同一个进程内的多个线程同一时刻只能有一个运行 即同一进程的多个线程无法实现并行=》无法利用多核优势 但是可以实现并发的效果 什么是多核优势? 多核即多个cpu,多个cpu带来的优势是计算性能的提升,所以 IO密集型: 多线程 计算密集型 多进程 GIL vs 自定义互斥锁 GIL相当于执行权限,意思是在一个进程内的多个线程想要执行,必须 先抢GIL,这把锁的特点是,当一个线程被剥夺走cpu的执行权限的同时会被 解释器强行释放GIL 2、进程池与线程池 什么是? 池:池用来限制进程/线程个数的一种机制 为何用? 当并发的任务数远大于计算机的承受能力,应该用池的概念 将并发的进程数/线程数控制在计算机能承受的数目 如何用? from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor def task(n): return n**2 def func(future): print(future.result()) if __name__ == '__main__': p=ProcessPoolExecutor(4) p.submit(task,10).add_done_callback(func) p.shutdown(wait=True) # pool.close() # pool.join() print('主') 提交任务的两种方式: 1、同步:提交完任务后就在原地等待,直到任务运行完毕并且拿到返回值后,才运行下一行代码 2、异步:提交完任务(绑定一个回调函数)后不原地等待,直接运行下一行代码,等到任务运行有返回值自动触发回调的函数的运行 程序的运行状态(阻塞,非阻塞) 1、阻塞: IO阻塞 2、非阻塞: 运行 就绪 今日内容: 1、协程 2、IO模型 阻塞IO 非阻塞IO IO多路复用 异步IO