浅谈python异步IO,同步IO,线程与进程~

【文章中代码未缩进，刚开始使用博客，后期会优化~】

主线程下的线程之间是可以通信的，但是父进程下的子进程之间不能主动通信，但是子进程想要实现通信也是可以的，可以选择折中的方法来实现，比如multiprocessing.Queue,用法与线程中的queue基本一致，直接上例子：
import threading
from multiprocessing import Process,Queue
import time
def thre(qq):
qq.put([1,'xixi',2])

if __name__ =='__main__':
q = Queue()
p = Process(target=thre,args=(q,))#进程中，因为进程内存是独立的，所以不能相互调用，必须传入参数，这个q其实是复制了一份Queue的实例，如果和线程一样不传参数，就会报错** not find。。。因为内存不共享。
#p =threading.Thread(target=thre) #线程中，不传参数是可以调用函数thre，因为他们是在同一个内存地址下操作【上面改为def thre():】，当然传参数也没问题。
p.start()
print(q.get())

#进程之间想要有联系，主动无法联系，这是硬伤，就如qq和word一样，但是如果非要他们有联系，就是从word复制文字到qq里（或者qq复制图片文字到word里面这样），这样貌似两者有联系，实际上只是克隆了那段文字的关系，但是看起来好像就有联系了，那么python中process之间的通信就是可以考虑通过 Queue来实现，Queue内部操作其实就是通过pickle的功能来实现传参数等各种联系的
还有一个是pipe：通过管道来传递，也是建立一个pipe的实例化对象。
from multiprocessing import Process,Pipe

def f(conn):
conn.send('balabala')
print(conn.recv())

if __name__=='__main__':
parent_conn,child_conn=Pipe()#实例化后是返回两个值，一个是父接头一个是子接头，因为是管道。
p = Process(target=f,args=(child_conn,))
p.start()
print(parent_conn.recv())
parent_conn.send('babababa')

这样也能实现数据的传递，但都不是共享
进程之间要实现共享，需要用manager。
from multiprocessing import Process,Manager
import time,os

def thre(dd,ll):
dd[os.getpid()] = os.getppid()
ll.append(os.getpid())
print(ll)
print(dd)

if __name__ =='__main__':
manager = Manager()
d = manager.dict()
l = manager.list(range(3))
t_list = []
for i in range(10):
p = Process(target=thre,args=(d,l))
p.start()
t_list.append(p)
for res in t_list:
res.join()
此时字典 d 和 列表 l，他们的数据同时都可以被进程修改覆盖，只不过我这里用的是os.getpid()获取的数据不一致，如果是一致的，那么最终字典只有一个k-v，列表是10个一样的数据。

进程锁的存在是为了输出同一个屏幕不要乱。。。仅此而已

进程池的作用和线程中的信号量差不多，同一时间允许几个进程同时运行
其中有 apply 和apply_async，一个是串行操作，一个是并行操作。
from multiprocessing import Pool
import time,os

def thre(dd):
time.sleep(1)
print('the process:',os.getpid())
return dd+100
def g(c):
print('haha',c,os.getpid())

#start_time = time.time()
# l=[]
if __name__ =='__main__':
p_ = Pool(3)#允许同时运行的进程数为3。
print(os.getpid())
for i in range(10):
p_.apply_async(func=thre,args=(i,),callback=g)【callback是回调函数，传的参数是thre的返回值】
p_.close()
p_.join()#这里如果不加join，在并行中会直接close，程序会直接关闭，加了join，主进程就会等待子进程结束以后最后才关闭，这个只在并行中有用，串行中没有什么作用。一定要先close再join

 

协程：可以实现高并发，本质上就是单线程，一个cpu支持上万个协程并发
gevent(自动触发) 和 greenlet(手动触发)
import gevent

def fun1():
print('runing 1 ...')
gevent.sleep(2)#模仿io
print('running 2 ...')

def fun2():
print('running 3 ...')
gevent.sleep(3)
print('running 4')

def fun3():
print('running 5 ...')
gevent.sleep(0)
print('end?')

gevent.joinall([gevent.spawn(fun1),gevent.spawn(fun2),gevent.spawn(fun3)])
运行结果：
runing 1 ...
running 3 ...
running 5 ...
end?
running 2 ...
running 4
----------------------
sleep相当于触发的按钮，出现一次sleep，就去找下一个函数中的内容打印等操作，sleep内的时间相当于他卡几次，sleep（3）相当于卡3秒，如果其他已经没卡着，就马上执行没卡着的语句，知道最后回来等到时间结束执行最后这个语句。协程用于多并发爬虫中效果很好。
import gevent,time
import urllib.request as ul
from gevent import monkey
monkey.patch_all()#这个标识代表把所有程序都当做io直接切换操作，不加这句话，因为gevent不会辨认出urllib的有io操作，相当于串行操作。
def f(url):
print('GET %s'%url)
res = ul.urlopen(url).read()
print('recv bytes %s from %s'%(len(res),url))

time_start = time.time()
l=['https://www.python.org/','http://km.58.com/','http://kan.sogou.com/dongman/','http://news.sohu.com/']
for i in l:
f(i)
print('同步时间：',time.time()-time_start)
async_time = time.time()
gevent.joinall([gevent.spawn(f,'https://www.python.org/'),
gevent.spawn(f,'http://km.58.com/'),
gevent.spawn(f,'http://kan.sogou.com/dongman/'),
gevent.spawn(f,'http://news.sohu.com/')])
print('异步时间：',time.time()-async_time)

运行结果：
GET https://www.python.org/
recv bytes 48860 from https://www.python.org/
GET http://km.58.com/
recv bytes 104670 from http://km.58.com/
GET http://kan.sogou.com/dongman/
recv bytes 12713 from http://kan.sogou.com/dongman/
GET http://news.sohu.com/
recv bytes 170935 from http://news.sohu.com/
同步时间： 3.780085563659668
GET https://www.python.org/
GET http://km.58.com/
GET http://kan.sogou.com/dongman/
GET http://news.sohu.com/
recv bytes 12690 from http://kan.sogou.com/dongman/
recv bytes 170935 from http://news.sohu.com/
recv bytes 104670 from http://km.58.com/
recv bytes 48860 from https://www.python.org/
异步时间： 2.5934762954711914

用户空间和内核空间(kernel)
现在操作系统中都是采用虚拟存储器，操作系统的核心是内核，独立于普通的应用程序，可以访问受保护的内存空间，也有访问硬件设备的权限，为了保证用户进程不能直接操作内核（kernel），保证内核的安全，操作系统把虚拟空间分为两部分，一部分为内核空间，一部分为用户空间。

进程切换
为了控制进程的执行，内核必须有能力挂起在CPU上运行的进程，并且恢复以前挂起的某个进程的执行，这种行为称作进程切换，因此，任何进程都是在操作系统内核的支持下运行的，与内核紧密相连。
从一个进程的运行转到另一个进程上运行，其实就是保存上下文就切换了。下次再来又从之前保存的位置开始。

进程的阻塞：
正式执行的进程，由于期待的某件事情并未发生，如请求系统资源失败等待，等待某种操作的完成，新数据尚未达到或无新工作开始等，则有系统自动执行阻塞原语，使自己由原来的运行状态转为阻塞状态暂停等待（）
。可见，进程的阻塞是进程自身的一种主动行为，也因此只有处于运行状态的进程（获得CPU），才可能将其转为阻塞状态，当进程进入阻塞状态时候，不耗费CPU资源的。

缓存I/O
又被成为标准IO，大多数文件系统默认I/O操作都是缓存I/O，在Linux的缓存I/O机制当中，操作系统会将I/O的数据缓存在文件系统的页缓存中，也就是说，文件数据会被拷贝到系统内核的缓冲区中，然后再从系统内核的缓冲区拷贝到用户的进程内存里也就是应用程序的地址空间。缺点就是数据会在用户进程应用程序地址空间和内核空间反复拷贝操作，这时对于CPU和内存的开销很大。

I/O模式
同步IO和异步IO：
同步IO中有：阻塞IO(blocking I/O),非阻塞IO(non-blocking I/O),多路复用IO(I/O multiplexing) 信号驱动（实际中不常用。在此暂时不记录笔记）
异步I/O(asynchronous I/O)

阻塞IO：发起请求，然后等待数据准备（此时进程阻塞等待），直到数据准备好接受时，又到内核空间开始copy给用户进程，此时又一次阻塞等待，直到数据全部发给用户进程（客户端）。

非阻塞IO：发起请求后，疯狂发送验证，数据未准备好时，并不会阻塞block，而是返回一个error给用户进程，用户进程会验证是否error，是就继续发出请求，来回验证，（此时由于进程没有阻塞，还可以干其他事，）不是就到了内核空间开始copy数据，此时其实还是阻塞，如果数据小会很快，数据大还是会感受到卡。最后用户收到完整数据。

多路复用I/O：一次发起几百次请求链接，无论哪条链接有数据回复，都会通知用户进程开始接受数据，此时那几条链接又开始进行内核copy直到进程收到完整数据（其实这里也是阻塞的）。这个模式的核心其实是用非阻塞IO的方式来驱动，所以形成多路复用，在用户看来已经是多并发了。

异步I/O：这个就牛逼了，他发起请求，当场就收到回复‘去干你其他的事’，此时该进程开始其他部分运行，并未有任何阻塞，收到数据时，直接后台开始内核copy，全部搞完以后直接‘送快递到家门口’，给一个信号通知，用户进程顺手就接受了数据，此时整个进程根本没有任何阻塞过程！这就是异步IO。

selectors
selectors中涵盖了select,poll,epoll,详细实例：
import selectors,socket
sel = selectors.DefaultSelector()

def accept(sock,mask):
conn,addr = sock.accept()
conn.setblocking(False)
sel.register(conn,selectors.EVENT_READ,read)

def read(conn,mask):
data = conn.recv(1024).decode()
if data:
conn.send(('haha+%s'%data).encode())
else:
print('发什么?',conn)
sel.unregister(conn)
conn.close()

sock = socket.socket()
sock.bind(('localhost',5000))
sock.listen(1000)
sel.register(sock,selectors.EVENT_READ,accept)
while True:
events= sel.select()
for key,mask in events:
callback = key.data
callback(key.fileobj,mask)
可以进行多并发运行。