python基本使用（3）— 多进程

本是古典何须时尚 2022-05-29 06:15 197阅读 0赞

1.多进程

是一种多核运算，Multiprocessing 和多线程threading 类似，它弥补了 threading 的一些劣势，比如 GIL。

1.1 添加进程 process + 存储进程输出 queue

创建方式与线程的类似

Queue的功能是将每个核或线程的运算结果放在队列中， 等到每个线程或核运行完毕后再从队列中取出结果， 继续加载运算。原因很简单, 多线程调用的函数不能有返回值, 所以使用Queue存储多个线程运算的结果

`args` 的参数只有一个值的时候，参数后面需要加一个逗号，表示`args`是可迭代的，后面可能还有别的参数，不加逗号会出错

import multiprocessing as mp
    
    def job(a):
        res=0
        for i in range(100):
            res+=i+i**2+i**3
        a.put(res)
    
    if __name__=='__main__':
        a=mp.Queue()    #定义一个多线程队列，用来存储结果
        p1=mp.Process(target=job,args=(a,)) #定义两个线程函数，用来处理同一个任务，注意args的参数
        p2=mp.Process(target=job,args=(a,))
        p1.start()
        p2.start()
        p1.join()
        p2.join()
        res1=a.get()
        res2=a.get()
        print(res1)
        print(res2)
        print(res1+res2)

结果为

RESTART: C:\Users\Administrator\AppData\Local\Programs\Python\Python36\4.py 
    24835800
    24835800
    49671600

1.2 多线程、多进程效率对比

import multiprocessing as mp
    import threading as td
    import time
    
    def job(q):
        res=0
        for i in range(1000000):
            res+=i+i**2+i**3
        q.put(res)   #queue
    
    def multicore():  
        q=mp.Queue()    #定义一个多进程队列，用来存储结果
        p1=mp.Process(target=job,args=(q,)) #定义两个线程函数，用来处理同一个任务，注意args的参数
        p2=mp.Process(target=job,args=(q,))
        p1.start()
        p2.start()
        p1.join()
        p2.join()
        res1=q.get()
        res2=q.get()
        print('multicore:',res1+res2)
    
    def multithread():     #多线程
        q = mp.Queue() # thread可放入process同样的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('multithread:', res1 + res2)
    
    def normal():    #普通函数
        res = 0
        for _ in range(2):
            for i in range(1000000):
                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)
        multicore()
        print('multicore time:', time.time() - st2)

结果为

RESTART: C:\Users\Administrator\AppData\Local\Programs\Python\Python36\4.py 
    normal: 499999666667166666000000
    normal time: 1.5049998760223389
    multithread: 499999666667166666000000
    multithread time: 1.5130000114440918
    multicore: 499999666667166666000000
    multicore time: 0.9110000133514404

我们发现多核/多进程最快，说明在同时间运行了多个任务。 而多线程的运行时间居然比什么都不做的程序还要慢一点，说明多线程还是有一定的短板的。

1.3 进程池

进程池`Pool`。 进程池就是我们将所要运行的东西，放到池子里，Python会自行解决多进程的问题。

1.  `Pool`默认调用是CPU的核数，传入`processes`参数可自定义CPU核数
2.  `map()` 放入迭代参数，返回多个结果
3.  `apply_async()`只能放入一组参数，并返回一个结果，如果想得到`map()`的效果需要通过迭代

import multiprocessing as mp
    
    
    def job(x):
        return x * x
    
    
    def multicore():
        pool = mp.Pool()  # 默认运行所有的核
        # 有了池子之后，就可以让池子对应某一个函数，我们向池子里丢数据，池子就会返回函数返回的值。
        # Pool和之前的Process的不同点是丢向Pool的函数有返回值，而Process的没有返回值。
    
        # 接下来用map()获取结果，在map()中需要放入函数和需要迭代运算的值，然后它会自动分配给CPU核，返回结果
        # pool = mp.Pool(processes=2)  #运行2个核
    
        res = pool.map(job, range(10))  # 1.map功能
        print(res)
    
        # Pool除了map()外，还有可以返回结果的方式，那就是apply_async().
        # apply_async()中只能传递一个值，它只会放入一个核进行运算，但是传入值时要注意是可迭代的，
        # 所以在传入值后需要加逗号,同时需要用get()方法获取返回值
    
        res = pool.apply_async(job, (2,))  # 2.apply_async功能，2的位置，只能输入一个值
        print(res.get())
    
        # 要想输入多个，用迭代器
        for i in range(10):
            multi_res = pool.apply_async(job, (i,))  # 这样的输出结果不是我们想要的
        # print(multi_res.get())
        print(res.get())
    
        # 要想输入多个，用迭代器
        # 可以看出在apply用迭代器的得到的结果和用map得到的结果是一样的
        # for i in range(10):
        multi_res = [pool.apply_async(job, (i,)) for i in range(10)]  # 3.
        # print(multi_res.get())
        print([res.get() for res in multi_res])  # 注意这里的两个for的范围
        # 在取出值时也需要一个一个取出来
    
    
    if __name__ == '__main__':
        multicore()

结果：

RESTART: C:\Users\Administrator\AppData\Local\Programs\Python\Python36\4.py 
    [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
    4
    4
    [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

参考：https://morvanzhou.github.io/tutorials/python-basic/multiprocessing/5-pool/

1.4 共享内存

共享变量可以通过 global 定义实现，但是，如果想要 cpu 之间的变量相互交流，则需要共享内存。

import multiprocessing as mp
    
    value1=mp.Value('i',2)
    value2=mp.Value('d',3.14)  #d表示双精度浮点型
    value3=mp.Array('i',[1,2,3])

这里的`Array`和numpy中的不同，它只能是一维的，不能是多维的。同样和`Value` 一样，需要定义数据形式，否则会报错。 我们会在后一节举例说明这两种的使用方法.

1.5 进程锁

import multiprocessing as mp
    import time
    
    def job(v, num, l):
        l.acquire() # 锁住  设置进程锁，保证运行时一个进程对锁内内容的独占
        for _ in range(5):
            time.sleep(0.1) 
            v.value += num # 获取共享内存
            print(v.value)
        l.release() # 释放
    
    def multicore():
        l = mp.Lock() # 定义一个进程锁，小写l
        v = mp.Value('i', 0) # 定义共享内存
        p1 = mp.Process(target=job, args=(v,1,l)) # 需要将lock传入
        p2 = mp.Process(target=job, args=(v,3,l)) 
        p1.start()
        p2.start()
        p1.join()
        p2.join()
    
    if __name__ == '__main__':
        multicore()
    
    '''
    3
    6
    9
    12
    15
    16
    17
    18
    19
    20
    '''