基础10 多进程、协程（multiprocessing、greenlet、gevent、gevent.monkey、select、selector）...

1.多进程实现方式（类似于多线程）

 1 import multiprocessing
 2 import time,threading
 3 
 4 def thread_run():#定义一个线程函数
 5     print("我是子线程%s" %threading.get_ident())  #threading.get_ident()函数获取当前线程的id
 6 def run(name):#定义一个进程函数
 7     time.sleep(1)
 8     print("hello,我是进程%s" %name)
 9     t = threading.Thread(target=thread_run(),)#在进程下运行子线程
10     t.start()
11 if __name__ == '__main__':  #注意：启多进程必须要加这个判断，否则会有意想不到的错误
12     for i in range(3):
13         p = multiprocessing.Process(target=run,args=('bob %s' %i,))
14         p.start()
15         p.join()
16 
17 输出：
18 hello,我是进程bob 0
19 我是子线程28176
20 hello,我是进程bob 1
21 我是子线程27016
22 hello,我是进程bob 2
23 我是子线程27516

2.进程都是由父进程创建和启动的

 1 from multiprocessing import Process
 2 import os
 3 
 4 
 5 def info(title):
 6     print(title)
 7     print('module name:', __name__)
 8     print('当前进程父进程id:', os.getppid())
 9     print('当前进程 id:', os.getpid())
10     print("\n\n")
11 
12 
13 def f(name):
14     info('\033[31;1mcalled from child process function f\033[0m')
15     print('hello', name)
16 
17 if __name__ == '__main__':
18     info('\033[32;1mmain process line\033[0m')
19     p = Process(target=f, args=('bob',)) #
20     p.start()
21 
22 输出：
23 main process line
24 module name: __main__
25 当前进程父进程id: 12468    #这是pycharm的进程号
26 当前进程 id: 25692           #这是第一个info函数的进程号
27 
28 
29 
30 called from child process function f
31 module name: __mp_main__
32 当前进程父进程id: 25692 #这是第一个info函数的进程号
33 当前进程 id: 27252        #这是f函数内调用的info函数的进程号
34  
35 
36 
37 hello bob

3.进程间是不共享内存的（不同于线程），即进程之间是不能直接交换信息的
可以用以下方式实现进程间通信，用Queue队列（线程中用queue）

 1 import threading
 2 from multiprocessing import Process ,Queue
 3 
 4 
 5 def f(zi_q):#定义一个子进程函数，它往Queue里放值
 6     zi_q.put([42,None,'hello'])
 7 
 8 if __name__ == '__main__':
 9     fu_q = Queue()                        #实例化父进程的一个队列Queue
10     zi_p = Process(target=f, args=(fu_q,))#将父进程的Queue作为参数传给子进程zi_p，这里父进程的Queue和子进程的Queue
11                                           #不是同一个Queue，这里fu_q作为参数传给子进程的时候，是将fu_q
12                                           #复制了一份传给了子进程，子进程将值赋给子进程的Queue后，函数内部
13                                           #又通过pickle形式将值传给了父进程的Queue（fu_q）
14     zi_p.start()
15     print(fu_q.get())  #父进程获取Queue内的值
16     zi_p.join()
17 
18 
19 输出：
20 [42, None, 'hello']

4.实现不同进程间通信的另外一种方法

（1）管道（Pipe()）,类似于socket

 1 from multiprocessing import Process, Pipe
 2 def f(conn):
 3     conn.send([42, 'hello'])#类似于socket，这里子进程发了两次，下边父进程就需要收两次
 4     conn.send([32, 'yyyy'])
 5     print(conn.recv())        #子进程也可以接收父进程发过来的信息
 6     conn.close()
 7 
 8 if __name__ == '__main__':
 9     parent_conn, child_conn = Pipe()
10     p = Process(target=f, args=(child_conn,))
11     p.start()
12     print(parent_conn.recv())#收两次
13     print(parent_conn.recv())
14     parent_conn.send('我是父进程A') #父进程可以给子进程发信息
15     p.join()
16 
17 输出：
18 [42, 'hello']
19 [32, 'yyyy']
20 我是父进程A

5.实现进程间内存的共享（所谓共享即所有进程操作同一份列表、字典等等，其实也不真的是同一份数据，只不过是程序内部通过复制原始列表、字典给进程，进程运行以后修改列表、字典，最后合为一个列表、字典，让人们以为是所有进程共享了相同的内存）
manager模块可实现这一功能

 1 from multiprocessing import Process, Manager
 2 import os
 3 def f(d, li):
 4     d[os.getpid()] = os.getpid()  #生成进程号字典
 5     li.append(os.getpid())         #生成进程号列表
 6     print(li)
 7 
 8 if __name__ == '__main__':
 9     with Manager() as manager:  #等同于manager = Manager()
10         d = manager.dict()   #创建一个字典,可在多个进程间共享和传递
11         li = manager.list()#创建一个列表,可在多个进程间共享和传递
12         p_list = []
13         for i  in range(5):#生成5个进程
14             p = Process(target=f, args=(d, li))
15             p.start()
16             p_list.append(p) #将这5个进程放进一个列表里
17         for res in p_list: #让每个进程都执行完
18             res.join()
19         print(d)
20 
21 输出：
22 [53812]
23 [53812, 51612]
24 [53812, 51612, 53520]
25 [53812, 51612, 53520, 53696]
26 [53812, 51612, 53520, 53696, 53608]
27 {53520: 53520, 51612: 51612, 53608: 53608, 53812: 53812, 53696: 53696}

6.进程锁，防止在屏幕上或日志里打印乱了
如这种乱：hello worhello world 8

 1 from multiprocessing import Process, Lock
 2 
 3 
 4 def f(l, i):
 5     l.acquire()#获得一把锁
 6     print('hello world', i)
 7     l.release()#释放一把锁
 8 if __name__ == '__main__':
 9     lock = Lock()
10     for num in range(10):
11         Process(target=f, args=(lock, num)).start()
12 
13 输出：
14 hello world 2
15 hello world 1
16 hello world 4
17 hello world 3
18 hello world 8
19 hello world 0
20 hello world 7
21 hello world 5
22 hello world 6
23 hello world 9

7.很重要进程池防止进程启动太多，消耗太多资源，一般是几个cpu就启几个进程

 1 from multiprocessing import Process, Pool
 2 import time
 3 import os
 4 
 5 def foo(i):
 6     time.sleep(1)
 7     print("正在执行子进程", os.getpid())
 8     return os.getpid()      #返回的是子进程的id号
 9 
10 def bar(arg):    #定义回调函数，参数arg是foo函数的返回值
11     print("子进程%s执行完毕,主进程%s调用的我"% (arg, os.getpid()))
12 
13 if __name__ == '__main__':
14     pool = Pool(5)  #允许进程池同时放入5个进程
15     print("主进程", os.getpid())
16     for i in range(10):
17         pool.apply_async(func=foo, args=(i,), callback=bar)#异步，异步最好，callback后跟回调函数，回调函数是由主进程执行的
18                                                         #回调函数的参数不用写，默认就是前边的func
19        # pool.apply(func=foo, args=(i,))  #同步，是串行的，看不出是5个5个的执行
20     print("主进程：end")
21     pool.close()        #注意，这里要先close，再join，否则会有错误
22     pool.join()

输出：

 1 主进程 57784
 2 主进程：end
 3 正在执行子进程 58060
 4 子进程58060执行完毕,主进程57784调用的我
 5 正在执行子进程 55428
 6 子进程55428执行完毕,主进程57784调用的我
 7 正在执行子进程 58356
 8 子进程58356执行完毕,主进程57784调用的我
 9 正在执行子进程 56716
10 子进程56716执行完毕,主进程57784调用的我
11 正在执行子进程 57380
12 子进程57380执行完毕,主进程57784调用的我
13 正在执行子进程 58060
14 子进程58060执行完毕,主进程57784调用的我
15 正在执行子进程 55428
16 子进程55428执行完毕,主进程57784调用的我
17 正在执行子进程 58356
18 子进程58356执行完毕,主进程57784调用的我
19 正在执行子进程 56716
20 子进程56716执行完毕,主进程57784调用的我
21 正在执行子进程 57380
22 子进程57380执行完毕,主进程57784调用的我

8.协程

又称微线程，是一种用户态的轻量级线程，协程拥有自己的寄存器和栈，协程调度切换时，将寄存器上下文和栈保存在其他地方，当切回来的时候，恢复先前保存的寄存器上下文和栈，因此，每一次协程恢复时，

都能在它离开中断的地方继续。

协程的好处：

无需线程上下文切换的开销
无需原子操作锁定及同步的开销
方便切换控制流，简化编程模型
高并发+高扩展性+低成本：一个CPU支持上万的协程都不是问题。所以很适合用于高并发处理。

缺点：

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

用greenlet模块实现协程（手动）：

 1 from greenlet import greenlet
 2 def test1():
 3     print("test1")      #4.打印
 4     gr2.switch()        #5.切换到test2，test2开始执行
 5     print("test1...")   #8.接上次离开的地方，开始执行，打印
 6     gr2.switch()        #9.切换到test2，test2开始执行
 7 
 8 def test2():
 9     print("test2")     #6.打印
10     gr1.switch()       #7.切换到tets1，test1开始执行
11     print("test2...")  #10.接上次离开的地方，开始执行，打印，程序结束
12 
13 gr1 = greenlet(test1)   #1.启动一个协程
14 gr2 = greenlet(test2)   #2.启动一个协程
15 gr1.switch()            #3.切换到test1，test1开始执行
16 
17 输出：
18 test1
19 test2
20 test1...
21 test2...

重要：用gevent模块实现协程（自动切换）。遇到IO（程序睡觉或卡主）便自动切换到下一个协程

 1 import gevent
 2 
 3 
 4 def t1():
 5     print("t1 运行")       #4.t1开始运行，打印
 6     gevent.sleep(2)       #5.切换到下一个协程，按生成的顺序，切换到t2
 7                           #10.程序运行到这里，只花费了不到0.1秒钟，这时t1还在睡觉，又触发了切换操作，切换到t2
 8                           #13.程序运行到这里，只花费了不到0.1秒钟，这时t1还在睡觉，又触发了切换操作，切换到t2
 9     print("再次回到t1")    #16.接上次离开的地方，执行打印操作
10 
11 
12 def t2():
13     print("t2 运行")       #6.t2开始运行，打印
14     gevent.sleep(1)       #7.切换到下一个协程，按生成的顺序，切换到t3
15                           #11.程序运行到这里，只花费了不到0.1秒钟，这时t2还在睡觉，又触发了切换操作，切换到t3
16                           #14.程序运行到这里，只花费了不到0.1秒钟，这时t2还在睡觉，又触发了切换操作，但是t3运行
17                           #    完毕，无法切换到t3，只能切换到t1，t1还在睡觉，又切换到t2，互相切换过了1秒后，t2睡觉
18                           #    完毕，这时切换到t2
19     print("再次回到t2")    #15.接上次离开的地方，执行打印操作，切换到t1
20 
21 
22 def t3():
23     print("t3运行")        #8.t3开始运行，打印
24     gevent.sleep(0)       #9.虽然睡觉0秒，但是会触发切换操作，切换到下一个协程，按生成的顺序，切换到t1
25     print("再次回到t3")    #12.由于t3没有睡觉，执行打印操作。t3运行完毕，自动切换到t1
26 
27 gevent.joinall([
28     gevent.spawn(t1),      #1.生成协程1
29     gevent.spawn(t2),      #2.生成协程2
30     gevent.spawn(t3),      #3.生成协程3
31 ])
32 
33 输出：
t1 运行
   t2 运行
   t3运行
   再次回到t3
   再次回到t2
   再次回到t1

9.爬虫，简单下载网页，使用gevent和gevent.monkey模块（需python3.5以上，gevent在linux完美支持，在Windows可能会有问题）

 1 from urllib import request
 2 import gevent,time
 3 from gevent import monkey
 4 monkey.patch_all() #把当前程序的所有的io操作给gevent单独的做上标记,使gevent能辨识出io操作，从而达到协程遇到io就切换的效果，不加monkey是不能识别出io的
 5 
 6 def f(url):
 7     print('GET: %s' % url)
 8     resp = request.urlopen(url)
 9     data = resp.read()
10     print('%d bytes received from %s.' % (len(data), url))
11 
12 urls = ['https://www.python.org/',
13         'https://www.yahoo.com/',
14         'https://github.com/' ]
15 time_start = time.time()
16 for url in urls:
17     f(url)
18 print("同步cost",time.time() - time_start)
19 async_time_start = time.time()
20 gevent.joinall([
21     gevent.spawn(f, 'https://www.python.org/'),
22     gevent.spawn(f, 'https://www.yahoo.com/'),
23     gevent.spawn(f, 'https://github.com/'),
24 ])
25 print("异步cost",time.time() - async_time_start)

输出：

 1 GET: https://www.python.org/
 2 47403 bytes received from https://www.python.org/.
 3 GET: https://www.yahoo.com/
 4 460629 bytes received from https://www.yahoo.com/.
 5 GET: https://github.com/
 6 25734 bytes received from https://github.com/.
 7 同步cost 5.380241632461548
 8 GET: https://www.python.org/
 9 GET: https://www.yahoo.com/
10 GET: https://github.com/
11 47403 bytes received from https://www.python.org/.
12 430893 bytes received from https://www.yahoo.com/.
13 25734 bytes received from https://github.com/.
14 异步cost 1.2949371337890625

10.基于协程的异步多并发

socket server

 1 #!usr/bin/env/python
 2 # -*- coding:utf-8 -*-
 3 # from  wangteng
 4 #通过协程实现并发
 5 import sys
 6 import socket
 7 import time
 8 import gevent
 9 
10 from gevent import socket,monkey
11 monkey.patch_all()
12 
13 
14 def server(port):
15     s = socket.socket()
16     s.bind(('0.0.0.0',port))
17     s.listen(500)#最大连接数为500
18     while True:
19         cli, addr = s.accept()
20         gevent.spawn(handle_request, cli)#  创建协程，这里做了一个操作，将cli作为参数传给handle_request,  handle_request(cli)
21 
22 
23 def handle_request(conn):
24     try:
25         while True:
26             data = conn.recv(1024)#接收client的请求
27 
28             with open("bbb.txt",'ab') as f:#接收文件内容，并写入文件
29                 f.write(data)
30             print("recv:", data)
31             # post = bytes('get your request: ',encoding='utf-8')
32             # post = post + data  #post为要向client发送的数据，必须为bytes类型
33             conn.send(data)  #向client发送数据
34             if not data:
35                 conn.shutdown(socket.SHUT_WR)
36 
37     except Exception as ex:
38         print(ex)
39 
40     finally:
41         conn.close()
42 
43 if __name__ == "__main__":
44     server(9000)

client

 1 #!usr/bin/env/python
 2 # -*- coding:utf-8 -*-
 3 # from  wangteng
 4 #本客户端可以跟异步socket_server多并发、select socket server一起用
 5 import socket
 6 
 7 host = 'localhost'
 8 port = 9000
 9 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
10 s.connect((host, port))
11 while True:
12     msg = bytes(input(">>:"), encoding="utf-8")  #输入要发送的内容，必须是bytes类型
13     s.sendall(msg)   #发送数据
14     # filename = input("filename:>>")   #注释的内容是可以发送文件的
15     # with open(filename,'rb') as f:#发送文件
16     #     for line in f:
17     #         s.sendall(line)
18     #         data = s.recv(1024)
19     #         print('recv:',data)
20     data = s.recv(1024)#接收数据
21     print('recv:', data.decode())#解码可以显示中文
22 s.close()

11.io多路复用，不是异步基于select
server

 1 #io多路复用实现并发，并不是异步  select
 2 
 3 import select
 4 import socket
 5 import queue
 6 
 7 
 8 server = socket.socket()
 9 server.bind(('localhost',9000))
10 server.listen(1000)
11 
12 
13 server.setblocking(False) #不阻塞
14 
15 msg_dic = {}
16 
17 inputs = [server, ]#新建链接的集合列表，实际情况是随着连接的增多，列表中的conn会越来越多，而server只有一个
18 #inputs = [server,conn] #[conn,]
19 #inputs = [server,conn,conn2,conn3...] #实际情况是随着连接的增多，列表中的conn会越来越多，而server只有一个
20 outputs = [] #要返回数据的客户端连接的集合列表，比如conn1需要server给conn1返回请求数据，就将conn1放进outputs里
21 #outputs = [r1,] #
22 while True:
23     readable, writeable, exceptional = select.select(inputs, outputs, inputs )#select监测新建立的连接,readble是inputs列表中
24                                                  #的元素集合，writeable是outputs列表中的元素集合，exceptional是异常连接的集合，它包含在inputs中
25     print(readable, writeable, exceptional)
26     for r in readable:
27         if r is server: #代表来了一个新连接
28             conn,addr = server.accept()#建立连接
29             print("来了个新连接",addr)
30             inputs.append(conn) #是因为这个新建立的连接还没发数据过来，现在就接收的话程序就报错了，
31             #所以要想实现这个客户端发数据来时server端能知道，就需要让select再监测这个conn，所以将新连接放到inputs里
32             msg_dic[conn] = queue.Queue() #初始化一个队列，用来存放要返回给这个客户端的数据
33         else: #conn2  这里不是新来了一个新连接，而是之前已经建立好的连接，是继续要发送数据的客户端连接
34             data = r.recv(1024)
35             print("收到数据",data.decode())
36             msg_dic[r].put(data)#将要返回给客户端的数据存放到队列里
37             outputs.append(r) #将连接r 放入要返回数据的连接队列里，如果r不是需要返回数据的连接，就可以不将r放进outputs中
38 
39     for w in writeable: #要返回给客户端的连接列表
40         data_to_client = msg_dic[w].get()#从队列里取出要返回给客户端的数据
41         w.send(data_to_client) #给客户端返回数据
42         outputs.remove(w) #确保下次循环writeable的时候,不返回这个已经处理完的连接了
43 
44     for e in exceptional:#有异常的客户端连接列表，称之为坏连接
45         if e in outputs: #如果坏连接在返回数据列表里，就将其去掉，否则会有错误
46             outputs.remove(e)
47 
48         inputs.remove(e) #坏连接一定在inputs中，所以不需要判断，因为所有连接都是最先存放在inputs中
49 
50         del msg_dic[e]  #同时，将队列中的坏连接也去掉（如果队列中存在的话）

client

 1 #!usr/bin/env/python
 2 # -*- coding:utf-8 -*-
 3 # from  wangteng
 4 #本客户端可以跟异步socket_server多并发、select socket server一起用
 5 import socket
 6 
 7 host = 'localhost'
 8 port = 9000
 9 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
10 s.connect((host, port))
11 while True:
12     msg = bytes(input(">>:"), encoding="utf-8")  #输入要发送的内容，必须是bytes类型
13     s.sendall(msg)   #发送数据
14     # filename = input("filename:>>")   #注释的内容是可以发送文件的
15     # with open(filename,'rb') as f:#发送文件
16     #     for line in f:
17     #         s.sendall(line)
18     #         data = s.recv(1024)
19     #         print('recv:',data)
20     data = s.recv(1024)#接收数据
21     print('recv:', data.decode())#解码可以显示中文
22 s.close()

12. 重要要会：IO多路复用之 selector
server

 1 #io多路复用实现并发，并不是异步  selector最好
 2 import selectors
 3 import socket
 4 
 5 sel = selectors.DefaultSelector()
 6 
 7 
 8 def accept(sock, mask):
 9     conn, addr = sock.accept()  # 新建一个连接
10     print('accepted', conn, 'from', addr,mask)
11     conn.setblocking(False)  #设置为非阻塞
12     sel.register(conn, selectors.EVENT_READ, read) #将新连接注册到sel实例中，由sel监测新连接conn，read为回调函数，即新连接conn将会调用read函数
13 
14 
15 def read(conn, mask):
16     data = conn.recv(1024)  # Should be ready
17     if data:
18         print('echoing', repr(data), 'to', conn)
19         conn.send(data)  # Hope it won't block
20     else:  #如果没有数据接收进来
21         print('closing', conn)
22         sel.unregister(conn)   #取消注册该连接conn
23         conn.close()
24 
25 
26 sock = socket.socket()
27 sock.bind(('localhost', 9000))
28 sock.listen(100)
29 sock.setblocking(False)
30 sel.register(sock, selectors.EVENT_READ, accept) #将sock注册到sel实例中，由sel监测sock，accept为回调函数，即sock要活动的话，
                                                    #将会调用accept函数去新建一个连接
31 
32 while True:
33     events = sel.select() #默认阻塞，有活动连接就返回活动的连接列表，events是一个列表，里边存放所有的连接
34     for key, mask in events:  #mask是连接的顺序数
35         callback = key.data #调用accept函数
36         callback(key.fileobj, mask) #key.fileobj= 文件句柄

client

 1 #!usr/bin/env/python
 2 # -*- coding:utf-8 -*-
 3 # from  wangteng
 4 #本客户端可以跟异步socket_server多并发、select socket server一起用
 5 import socket
 6 
 7 host = 'localhost'
 8 port = 9000
 9 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
10 s.connect((host, port))
11 while True:
12     msg = bytes(input(">>:"), encoding="utf-8")  #输入要发送的内容，必须是bytes类型
13     s.sendall(msg)   #发送数据
14     # filename = input("filename:>>")   #注释的内容是可以发送文件的
15     # with open(filename,'rb') as f:#发送文件
16     #     for line in f:
17     #         s.sendall(line)
18     #         data = s.recv(1024)
19     #         print('recv:',data)
20     data = s.recv(1024)#接收数据
21     print('recv:', data.decode())#解码可以显示中文
22 s.close()