python学习笔记-（十四）进程&协程

一. 进程

1. 多进程multiprocessing

multiprocessing包是Python中的多进程管理包，是一个跨平台版本的多进程模块。与threading.Thread类似，它可以利用multiprocessing.Process对象来创建一个进程。该进程可以运行在Python程序内部编写的函数。该Process对象与Thread对象的用法类似。

创建一个Process实例，可用start()方法启动。

join()方法可以等待子进程结束后再继续往下运行，通常用于进程间的同步。

1 2 3 4 5 6 7 8 9 10

from multiprocessing import Process import time def f(name):     time.sleep(2)     print('hello', name)    if __name__ == '__main__':     p = Process(target=f, args=('bob',))     p.start()     p.join()

写个程序，对比下主进程和子进程的ID：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15

from multiprocessing import Process import os def info(title):     print(title)     print('进程名称:', __name__)     print('父进程ID:', os.getppid())     print('子进程ID:', os.getpid())     print(" ") def f(name):     info('33[31;1mcalled from child process function f33[0m')     print('hello', name) if __name__ == '__main__':     info('33[32;1mmain process line33[0m')     p = Process(target=f, args=('bob',))     p.start()

2. 进程间通信

不同进程间内存是不共享的，要想实现两个进程间的数据交换，可以使用Queue、Pipe、Manager，其中：

1）Queue  Pipe 只是实现进程间数据的传递；

2）Manager 实现了进程间数据的共享，即多个进程可以修改同一份数据；

2.1 Queue

Queue允许多个进程放入，多个进程从队列取出对象，先进先出。（使用方法跟threading里的queue差不多）

1 2 3 4 5 6 7 8 9 10 11 12

from multiprocessing import Process,Queue def f(qq):     qq.put([42,None,"hello"])     qq.put([43,None,"HI"])   if __name__ == '__main__':     q = Queue()     p = Process(target=f,args=(q,))     p.start()     print(q.get())     print(q.get())     p.join()

2.2 Pipe

Pipe也是先进先出

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16

from multiprocessing import Process, Pipe   def f(conn):     conn.send([42, None, '儿子发送的消息'])     conn.send([42, None, '儿子又发消息啦'])     print("接收父亲的消息:",conn.recv())     conn.close()   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(parent_conn.recv())  # prints "[42, None, 'hello']"     parent_conn.send("回家吃饭！") # prints "[42, None, 'hello']"     p.join()

2.3 Manager

Manager对象类似于服务器与客户之间的通信 (server-client)，与我们在Internet上的活动很类似。我们用一个进程作为服务器，建立Manager来真正存放资源。其它的进程可以通过参数传递或者根据地址来访问Manager，建立连接后，操作服务器上的资源。在防火墙允许的情况下，我们完全可以将Manager运用于多计算机，从而模仿了一个真实的网络情境。

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18

from multiprocessing import Process,Manager import  os def f(d,l):     d[os.getpid()] = os.getpid()     l.append(os.getpid())     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()

3. 进程池

进程池 (Process Pool)可以创建多个进程。这些进程就像是随时待命的士兵，准备执行任务(程序)。一个进程池中可以容纳多个待命的士兵。

进程池有两种方法：

1）串行：apply

2）并行：apply_async

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18

from multiprocessing import Process,Pool import time import os 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(processes=3)#允许进程池同时放入3个进程     print("主进程",os.getpid())     for i in range(10):         pool.apply_async(func=Foo,args=(i,),callback=Bar)     print('end')     pool.close()     pool.join()#进程池中进程执行完毕后在关闭；如果注释则程序直接关闭

使用回调函数的目的是：在父进程中执行可以提高效率；（比如连接数据库，写回调函数的话，父进程连接一次数据库即可；如果使用子进程，则需要连接多次）

4. 其他(lock)

lock：屏幕上打印的锁，防止打印显示混乱

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16

from multiprocessing import Process, Lock 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。

协程拥有自己的寄存器上下文和栈。协程调度切换时，将寄存器上下文和栈保存到其他地方，在切回来的时候，恢复先前保存的寄存器上下文和栈。因此：协程能保留上一次调用时的状态（即所有局部状态的一个特定组合），每次过程重入时，就相当于进入上一次调用的状态，换种说法：进入上一次离开时所处逻辑流的位置。

好处：

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

缺点：

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

1.实例

传统的生产者-消费者模型是一个线程写消息，一个线程取消息，通过锁机制控制队列和等待，但一不小心就可能死锁。

如果改用协程，生产者生产消息后，直接通过yield跳转到消费者开始执行，待消费者执行完毕后，切换回生产者继续生产，效率极高。

代码示例：

+ View Code

输出结果：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15

[生产者] Producing 1... [消费者] Consuming 1... [生产者] 消费者返回状态码: 200 OK [生产者] Producing 2... [消费者] Consuming 2... [生产者] 消费者返回状态码: 200 OK [生产者] Producing 3... [消费者] Consuming 3... [生产者] 消费者返回状态码: 200 OK [生产者] Producing 4... [消费者] Consuming 4... [生产者] 消费者返回状态码: 200 OK [生产者] Producing 5... [消费者] Consuming 5... [生产者] 消费者返回状态码: 200 OK

注意到consumer函数是一个generator，把一个consumer传入produce后：

1. 首先调用c.send(None)启动生成器；
2. 然后，一旦生产了东西，通过c.send(n)切换到consumer执行；
3. consumer通过yield拿到消息，处理，又通过yield把结果传回；
4. produce拿到consumer处理的结果，继续生产下一条消息；
5. produce决定不生产了，通过c.close()关闭consumer，整个过程结束。

整个流程无锁，由一个线程执行，生产者和消费者协作完成任务，所以称为“协程”，而非线程的抢占式多任务。（原理：遇到I/O操作就切换，只剩下CPU操作（CPU操作非常快））

一句话总结协程的特点：子程序就是协程的一种特例。

python中支持协程的有以下两个模块：greenlet和greent

2. Greenlet

greenlet封装好的协程,利用.swith对协程操作进行手动切换

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17

from greenlet import  greenlet def test1():     print(12)     gr3.switch()     print(34)     gr2.switch()     print(78) def test2():     print(56)     gr1.switch() def test3():     print(90)     gr1.switch() gr1 = greenlet(test1)#启动协程 gr2 = greenlet(test2) gr3 = greenlet(test3) gr1.switch()

3. Greent

Gevent 是一个第三方库，可以轻松通过gevent实现并发同步或异步编程，在gevent中用到的主要模式是Greenlet, 它是以C扩展模块形式接入Python的轻量级协程。 Greenlet全部运行在主程序操作系统进程的内部，但它们被协作式地调度。

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18

import gevent def foo():     print("运行foo")     gevent.sleep(2)     print("再次回到foo") def bar():     print("这里是bar")     gevent.sleep(1)     print("又回到了bar") def func3():     print("运行func3")     gevent.sleep(0)     print("再次运行func3") gevent.joinall([     gevent.spawn(foo),     gevent.spawn(bar),     gevent.spawn(func3) ])

同步与异步的性能区别：

1）同步：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22

from gevent import monkey;   # monkey.patch_all() import gevent from  urllib.request import urlopen import time   def f(url):     print('GET: %s' % url)     resp = urlopen(url)     data = resp.read()     print('%d bytes received from %s.' % (len(data), url))   urls = [ 'https://www.python.org/',          'https://www.yahoo.com/',          'https://github.com/'          ]   time_start = time.time() for url in urls:     f(url) print("同步cost",time.time() - time_start)

2）异步：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24

from gevent import monkey;   # monkey.patch_all() import gevent from  urllib.request import urlopen import time   def f(url):     print('GET: %s' % url)     resp = urlopen(url)     data = resp.read()     print('%d bytes received from %s.' % (len(data), url))   urls = [ 'https://www.python.org/',          'https://www.yahoo.com/',          'https://github.com/'          ] async_time_start = time.time() gevent.joinall([     gevent.spawn(f, 'https://www.python.org/'),     gevent.spawn(f, 'https://www.yahoo.com/'),     gevent.spawn(f, 'https://github.com/'), ]) print("异步cost",time.time()-async_time_start )

结论：同步开销时间为4秒，异步开销为2.5秒，大大节省了开销，这就是协程的魅力；monkey.patch_all()使gevent能识别到urllib中的I/O操作

使用gevent实现单线程下的多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()