2020-01-18 •

Python多线程 N部曲(3)

第三曲

多进程

库名multiprocessing

简单例子

# 使用方法和多线程类似
from multiprocessing import Process
import os
import time

def P_func(name):
    print(name,os.getpid())    # 打印进程的 ID
    time.sleep(2)

p_list = []
if __name__ == '__main__':
    for i in range(10):
        p = Process(target=P_func,args=('process' + str(i),))
        p.start()
        p_list.append(p)

    for p in p_list:
        p.join()

ps: windows上没有语句if __name__ == '__main__':会报错

进程之间的通讯

— Pipe

# multiprocessing 里的 Queue 可以再进程与进程之间共享
# Pipe 相当于在两个进程之间建立一个通道实现通信
# Lock 的使用可以控制同一时间只能有一个进程对屏幕进行打印操作，防止输出内容混乱

from multiprocessing import Pipe,Process,Lock

def one_end_pipe(p,lock):
    p.send("hello from one_end_pipe")
    lock.acquire()
    print(p.recv())
    lock.release()
def the_other_end_pipe(p,lock):
     p.send("hello from the_other_end_pipe")
     lock.acquire()
     print(p.recv())
     lock.release()
if __name__ == '__main__':
    lock = Lock()
    one_end , the_other_end = Pipe()   # 返回连接的两端对象
    p1 = Process(target=one_end_pipe,args=(one_end,lock,))
    p2 = Process(target=the_other_end_pipe,args=(the_other_end,lock,))
    p1.start()
    p2.start()
    p1.join()
    p2.join()

— Queue

多进程里不能向多线程里一样多个线程去操作由queue 模块生成的 queue 对象。各个进程之间是独立占有各自的内存空间的。如果要用Queue 来对多进程实现控制，那么要导入multiprocessing 模块里的 Queue 类，该类的实例化对象可以用于多进程的控制操作。

from multiprocessing import Queue,Process
import os

def say_hello(name,q):
    print('get:',q.get(),'pid:',os.getpid())
    # q.put('in child process')

if __name__ == '__main__':
    q = Queue()    # 实例化一个适用于多进程之间的 queue
    q.put(1)
    q.put(2)
    q.put(3)
    q.put(4)
    q.put(5)
    q.put(6)
    q.put(7)
    q.put(8)

    p1 = Process(target=say_hello,args=('Binary',q,))
    p2 = Process(target=say_hello,args=('Joker',q,))
    p3 = Process(target=say_hello,args=('Bob',q,))
    p4 = Process(target=say_hello,args=('Joe',q,))
    p5 = Process(target=say_hello,args=('new_one',q,))
    p1.start()
    p2.start()
    p3.start()
    p4.start()
    p5.start()
    p1.join()
    p2.join()
    p3.join()
    p4.join()
    p5.join()

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

ps: 可以看到，每一个进程传入了一个 q，但是并非真正的将q 传了过去，而是在新的进程内存空间里再生了一个q ，每一个进程对q 的操作会通过序列化q的内容来告诉其他的进程。

进程池

控制同一时间有多少进程在cpu上执行

from multiprocessing import Pool
import os,time
def first_func(name):
    print('in first func:',name,os.getpid())
    time.sleep(2)

if __name__ == '__main__':
    pool = Pool(4)
    print('main process id:',os.getpid())
    for i in range(10):
        # pool.apply(first_func,(i,))         # 相当于串行
        pool.apply_async(first_func,args=('process'+str(i),))  # 进程池里的进程并行
    pool.close()
    pool.join()

进程之间的资源共享

— manager

import multiprocessing,os,time

def op(test_dic,test_list):
    test_dic[1] = 'dict in op'
    test_list.append(os.getpid())
    print('finish op ...')

if __name__ =='__main__':
    m = multiprocessing.Manager()
    m_dict = m.dict()           # 创建一个进程之间可共享的字典
    m_list = m.list()          # 创建一个进程之间可共享的列表
    for i in range(3):
        p = multiprocessing.Process(target=op,args=(m_dict,m_list,))
        p.start()
    time.sleep(5)
    print(m_dict)
    print(m_list)

ps: 通过manager可以创建的共享数据类型：list、dict、Namespace、Lock、RLock、Semaphore、BoundedSemaphore、Condition、Event、Barrier、Queue、Value、Array

Binary