python学习笔记-(十四)进程&协程

时间:2021-12-05 06:54:53

一. 进程

1. 多进程multiprocessing

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

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

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

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:

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

2. 进程间通信

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

1)Queue  \ Pipe 只是实现进程间数据的传递;

2)Manager 实现了进程间数据的共享,即多个进程可以修改同一份数据;

2.1 Queue

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

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也是先进先出

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运用于多计算机,从而模仿了一个真实的网络情境。

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

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:屏幕上打印的锁,防止打印显示混乱

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跳转到消费者开始执行,待消费者执行完毕后,切换回生产者继续生产,效率极高。

代码示例:

def consumer():
    r = ''
    while True:
        n = yield r
        if not n:
            return
        print('[消费者] Consuming %s...' % n)
        r = '200 OK'
def produce(c):
    c.send(None)
    n = 0
    while n < 5:
        n = n + 1
        print('[生产者] Producing %s...' % n)
        r = c.send(n)
        print('[生产者] 消费者返回状态码: %s' % r)
    c.close()

c = consumer()
produce(c)

输出结果:

[生产者] 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对协程操作进行手动切换

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全部运行在主程序操作系统进程的内部,但它们被协作式地调度。

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)同步:

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)异步:

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并发:

python学习笔记-(十四)进程&协程python学习笔记-(十四)进程&协程
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)
server端
python学习笔记-(十四)进程&协程python学习笔记-(十四)进程&协程
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()
client端