2.5.4.2Python-多进程_mp.pool

总目录：https://blog.csdn.net/qq_41106844/article/details/105553392

Python - 子目录：https://blog.csdn.net/qq_41106844/article/details/105553333

 

 

创建进程

当然从最开始创建一个线程开始

import multiprocessingas mp

def job(a,b):

    print('aaa')

 

if __name__ =='__main__':

    p1 = mp.Process(target=job,args=(1,2))

    p1.start()

    p1.join()

aaa

建立进程的过程和建立线程差不多。

进程队列

import multiprocessing as mp

 

def job(q):

    res =0

        for iin range(1000):

            res += i+i**2+i**3

    q.put(res)

 

if __name__ =='__main__':

    q = mp.Queue()

    p1 = mp.Process(target=job,args=(q,))#可迭代类型警告

    p2 = mp.Process(target=job, args=(q,))

    p1.start()

    p2.start()

    p1.join()

    p2.join()

    res1 = q.get()

    res2 = q.get()

    print(res1+res2)

队列的操作也和多线程一样。

 

进程池（pool）

如果是上百甚至上千个目标，手动的去创建进程的工作量巨大，此时就可以用到multiprocessing模块提供的Pool方法。

import multiprocessing as mp

 

def job(x):

    return x**2

def multicore():

    pool = mp.Pool()

    res = pool.map(job,range(10))

    print(res)

    res = pool.apply_async(job,(4,))

    print(res.get())

    multi_res=[pool.apply_async(job,(i,))for iin range(10)]

    print([res.get()for resin multi_res ])

if __name__ =='__main__':

    multicore()

[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

16 

[0, 1, 4, 9, 16, 25, 36, 49, 64, 81] 

我们来看一下pool（），map（）和apply_async（）三个方法，同时了解一下其他常用方法。

pool():进程池方法，使用其他方法必须在其实例化对象上使用。

 

apply_async( func,args = () , callback = None)  ：异步的效率,也就是池中的进程一次性都去执行任务.

    func : 进程池中的进程执行的任务函数

    args : 可迭代对象性的参数,是传给任务函数的参数

    callback : 回调函数,就是每当进程池中有进程处理完任务了,返回的结果可以交给回调函数,由回调函数进行进一步处理,回调函数只异步才有,同步没有.回调函数是父进程调用.

 

map( func,iterable)

    func : 进程池中的进程执行的任务函数

    iterable : 可迭代对象,是把可迭代对象那个中的每个元素一次传给任务函数当参数.

    map方法自带close和join

其他的方法：

apply ( func,args = ()) : 同步的效率,也就是说池中的进程一个一个的去执行任务

    func : 进程池中的进程要执行的任务函数

    args : 可迭代对象性参数,是传给任务函数的参数

同步处理任务时,不需要close和join,并且进程池中的所有进程都是普通进程。

 

close()：关闭Pool，使其不再接受新的任务；

terminate()：不管任务是否完成，立即终止；

join()：主进程阻塞，等待子进程的退出， 必须在close或terminate之后使用；

效率比较

我们来比较一下普通进程，多进程，多线程计算时运行的速度快慢

import multiprocessing as mp

import threading as td

import time

 

def job(q):

    res =0

        for iin range(10000):

            res += i+i**2+i**3

    q.put(res)

 

def multcore():

    q = mp.Queue()    

    p1 = mp.Process(target=job,args=(q,))#可迭代类型警告

    p2 = mp.Process(target=job, args=(q,))

    p1.start()

    p2.start()

    p1.join()

    p2.join()

    res1 = q.get()

    res2 = q.get()    

    print(res1+res2)

 

def multithread():

    q = mp.Queue()

    t1 = td.Thread(target=job,args=(q,))#可迭代类型警告

    t2 = td.Thread(target=job, args=(q,))

    t1.start()

    t2.start()

    t1.join()

    t2.join()

    res1 = q.get()

    res2 = q.get()

    print(res1+res2)

 

def normal():

    res =0

    for _in range(2):

        for iin range(10000):

            res += i + i**2 + i**3

    print("normal",res)

 

if __name__ =='__main__':

    st = time.time()

    normal()

    st1 = time.time()

    print("normal time",st1-st)    

    multithread()

    st2 = time.time()

    print("multithread time",st2-st1)    

    multcore()

    print("multicore time",time.time()-st2)

normal 4999666716660000 

normal time 0.016955137252807617 

4999666716660000

multithread time 0.02589702606201172 

4999666716660000

multicore time 0.47074174880981445     

 

共享内存与进程锁

共享内存（Shared Memory）是最简单的进程间通信方式，它允许多个进程访问相同的内存，一个进程改变其中的数据后，其他的进程都可以看到数据的变化。

Python中的共享内存就不得不提两个方法

value = mp.Value('d',1)

array = mp.Array('i',[1,3,5])

他们一个可以接收普通对象的数据，一个用来接收可迭代对象的数据。

我们来一个一个例子：

import multiprocessing as mp
import time

 

def job(v,num):

    for _ in range(10):

        time.sleep(0.1)

        v.value += num

        print(v.value)

 

def multcore():

    v = mp.Value('i',0)

    p1 = mp.Process(target=job,args=(v,1))

    p2 = mp.Process(target=job, args=(v,3))

    p1.start()

    p2.start()

    p1.join()

    p2.join()

 

if __name__ =='__main__':

    multcore()

我们可以看到两个进程都对v进行了修改，说明v在两个进程中都存在。

但是这样太乱了，进程a来一下，进程b来一下，我们可不可以让a执行完了再执行b呢。

我们就来说说进程锁：

import multiprocessing as mp

import time

 

def job(v,num,l):

    l.acquire()

    for _in range(10):

        time.sleep(0.1)

        v.value += num

        print(v.value)

    l.release()

 

def multcore():

    l = mp.Lock()

    v = mp.Value('i',0)

    p1 = mp.Process(target=job,args=(v,1,l))

    p2 = mp.Process(target=job, args=(v,3,l))

    p1.start()

    p2.start()

    p1.join()

    p2.join()

 

if __name__ =='__main__':

    multcore()

这样就整洁了一些。