Python系列之 - multiprocessing

Posted 2020-10-31 慕沁

由于GIL的存在，python中的多线程其实并不是真正的多线程，如果想要充分地使用多核CPU的资源，在python中大部分情况需要使用多进程。

multiprocessing包是Python中的多进程管理包。与threading.Thread类似，它可以利用multiprocessing.Process对象来创建一个进程。该进程可以运行在Python程序内部编写的函数。该Process对象与Thread对象的用法相同，也有start(), run(), join()的方法。此外multiprocessing包中也有Lock/Event/Semaphore/Condition类 (这些对象可以像多线程那样，通过参数传递给各个进程)，用以同步进程，其用法与threading包中的同名类一致。所以，multiprocessing的很大一部份与threading使用同一套API，只不过换到了多进程的情境。

调用方式1：
from multiprocessing import Process
import time
def f(name):
    time.sleep(1)
    print(\'hello\', name,time.ctime())

if __name__ == \'__main__\':
    p_list=[]
    for i in range(3):
        p = Process(target=f, args=(\'alvin\',))
        p_list.append(p)
        p.start()
　　　　　# p.terminate()# 终止进程　
    for i in p_list:
        p.join()
    print(\'end\')

调用方式2：类继承
from multiprocessing import Process
import time

class MyProcess(Process):
    def __init__(self):
        super(MyProcess, self).__init__()
        #self.name = name

    def run(self):
        time.sleep(1)
        print (\'hello\', self.name,time.ctime())


if __name__ == \'__main__\':
    p_list=[]
    for i in range(3):
        p = MyProcess()
        p.start()
        p_list.append(p)

    for p in p_list:
        p.join()

    print(\'end\')

为了显示涉及的单个进程id，这里有一个扩展示例:

from multiprocessing import Process
import os
import time

def info(title):
    print("title:",title)
    print(\'parent process:\',os.getppid())
    print(\'process id:\',os.getpid())
if __name__ == \'__main__\':
    info(\'main process line\')
    time.sleep(3)
    print("------------------")
    p = Process(target=info,args=(\'yuan\',))
    p.start()
    p.join()

结果：
title: main process line
parent process: 3504
process id: 4072
------------------
title: yuan
parent process: 4072
process id: 6024

  

 

构造方法：

Process([group [, target [, name [, args [, kwargs]]]]])

　　group: 线程组，目前还没有实现，库引用中提示必须是None； 
　　target: 要执行的方法； 
　　name: 进程名； 
　　args/kwargs: 要传入方法的参数。

实例方法：

　　is_alive()：返回进程是否在运行。

　　join([timeout])：阻塞当前上下文环境的进程程，直到调用此方法的进程终止或到达指定的timeout（可选参数）。

　　start()：进程准备就绪，等待CPU调度

　　run()：strat()调用run方法，如果实例进程时未制定传入target，这star执行t默认run()方法。

　　terminate()：不管任务是否完成，立即停止工作进程

属性：

　　daemon：和线程的setDeamon功能一样

　　name：进程名字。

　　pid：进程号。

Process类方法

  

进程间通讯

1. 进程队列Queue
2. 管道
3. Manangers

 进程对列Queue

from multiprocessing import Process, Queue
import queue

def f(q,n):
    #q.put([123, 456, \'hello\'])
    q.put(n*n+1)
    print("son process",id(q))

if __name__ == \'__main__\':
    q = Queue()  #try: q=queue.Queue()
    print("main process",id(q))

    for i in range(3):
        p = Process(target=f, args=(q,i))
        p.start()

    print(q.get())
    print(q.get())
    print(q.get())

 

管道

from multiprocessing import Process, Pipe

def f(conn):
    conn.send([12, {"name":"yuan"}, \'hello\'])
    response=conn.recv()
    print("response",response)
    conn.close()
    print("q_ID2:",id(conn))

if __name__ == \'__main__\':

    parent_conn, child_conn = Pipe()

   print("q_ID :",id(parent_conn))

print("q_ID1:",id(child_conn))
    p = Process(target=f, args=(child_conn,))
    p.start()
    print(parent_conn.recv())   # prints "[42, None, \'hello\']"
    parent_conn.send("儿子你好!")
    p.join()

Managers

Queue和pipe只是实现了数据交互，并没实现数据共享，即一个进程去更改另一个进程的数据。

from multiprocessing import Process, Manager

def f(d, l,n):
    d[n] = \'1\'
    d[\'2\'] = 2
    d[0.25] = None
    l.append(n)
    #print(l)

    print("son process:",id(d),id(l))

if __name__ == \'__main__\':

    with Manager() as manager:

        d = manager.dict()

        l = manager.list(range(5))

        print("main process:",id(d),id(l))

        p_list = []

        for i in range(10):
            p = Process(target=f, args=(d,l,i))
            p.start()
            p_list.append(p)

        for res in p_list:
            res.join()

        print(d)
        print(l)