玩转python进程池的使用与进程之间通信的实现

Bertha 。 2021-12-15 03:59 332阅读 0赞

**1.用queue实现两个进程之间数据传输，一个读取，一个写入**

from multiprocessing import Process, Queue
    import os, time, random
    
    # 写数据进程执行的代码:
    def write(q):
        for value in ['A', 'B', 'C']:
            print('Put %s to queue...' % value)
            q.put(value)
            time.sleep(random.random())
    
    # 读数据进程执行的代码:
    def read(q):
        while True:
            if not q.empty():
                value = q.get(True)
                print('Get %s from queue.' % value)
                time.sleep(random.random())
            else:
                break
    
    if __name__=='__main__':
        # 父进程创建Queue，并传给各个子进程：
        q = Queue()
        pw = Process(target=write, args=(q,))
        pr = Process(target=read, args=(q,))
        # 启动子进程pw，写入:
        pw.start()
        # 等待pw结束:
        pw.join()
        # 启动子进程pr，读取:
        pr.start()
        pr.join()
        # pr进程里是死循环，无法等待其结束，只能强行终止:
        print('')
        print('所有数据都写入并且读完')
    '''
    Put A to queue...
    Put B to queue...
    Put C to queue...
    Get A from queue.
    Get B from queue.
    Get C from queue.
    
    所有数据都写入并且读完
    '''

###   2.python进程池原理与实现演示 ###

**当需要创建的子进程数量不多时，可以直接利用multiprocessing中的Process动态成生多个进程，但如果是上百甚至上千个目标，手动的去创建进程的工作量巨大，此时就可以用到multiprocessing模块提供的Pool方法。**

**       初始化Pool时，可以指定一个最大进程数，当有新的请求提交到Pool中时，如果池还没有满，那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到指定的最大值，那么该请求就会等待，直到池中有进程结束，才会用之前的进程来执行新的任务**

**multiprocessing.Pool常用函数解析：**

*  apply\_async(func\[, args\[, kwds\]\]) ：使用非阻塞方式调用func（并行执行，堵塞方式必须等待上一个进程退出才能执行下一个进程），args为传递给func的参数列表，kwds为传递给func的关键字参数列表；
 *  close()：关闭Pool，使其不再接受新的任务；
 *  terminate()：不管任务是否完成，立即终止；
 *  join()：主进程阻塞，等待子进程的退出， 必须在close或terminate之后使用；

from multiprocessing import  Pool
    import  time ,os ,random
    
    def worker(msg):
        t_start = time.time() #获取当前系统时间，长整型，常用来测试程序执行时间
        print("%s开始执行,进程号为%d" % (msg,os.getpid()))
        # random.random()随机生成0~1之间的浮点数
        time.sleep(random.random()*2)
        t_stop = time.time()
        print(msg,"执行完毕，耗时%0.2f" % (t_stop-t_start))
    
    def main():
        #Pool的默认大小是CPU的核数
        po  = Pool(3)# 定义一个进程池，最大进程数3，大小可以自己设置，也可写成processes=3
        for i in range(0,10):
            # Pool().apply_async(要调用的目标,(传递给目标的参数元祖,))
            # 每次循环将会用空闲出来的子进程去调用目标
            po.apply_async(worker,(i,))
    
        print("----start----")
        po.close()  # 关闭进程池，关闭后po不再接收新的请求
        po.join()  # 等待po中所有子进程执行完成，必须放在close语句之后
        print("-----end-----")
    if __name__ == '__main__':  
        main()
    '''
    ----start----
    0开始执行,进程号为5056
    1开始执行,进程号为968
    2开始执行,进程号为5448
    2 执行完毕，耗时0.38
    3开始执行,进程号为5448
    1 执行完毕，耗时0.47
    4开始执行,进程号为968
    4 执行完毕，耗时0.02
    5开始执行,进程号为968
    3 执行完毕，耗时0.13
    6开始执行,进程号为5448
    5 执行完毕，耗时1.44
    7开始执行,进程号为968
    6 执行完毕，耗时1.45
    8开始执行,进程号为5448
    0 执行完毕，耗时1.99
    9开始执行,进程号为5056
    8 执行完毕，耗时0.18
    7 执行完毕，耗时0.58
    9 执行完毕，耗时1.75
    -----end-----'''

### 3.python中进程池中进程之间的通信Queue ###

** 注意使用的是multiprocessing.Manager()中的Queue()，而不是multiprocessing.Queue()**

# 修改import中的Queue为Manager
    from multiprocessing import Manager,Pool
    import os,time,random
    
    def reader(q):
        print("reader启动(%s),父进程为(%s)" % (os.getpid(), os.getppid()))
        for i in range(q.qsize()):
            print("reader从Queue获取到消息：%s" % q.get(True))
    
    def writer(q):
        print("writer启动(%s),父进程为(%s)" % (os.getpid(), os.getppid()))
        for i in "itcast":
            q.put(i)
    
    if __name__=="__main__":
        print("(%s) start" % os.getpid())
        q = Manager().Queue()  # 使用Manager中的Queue
        po = Pool()
        po.apply_async(writer, (q,))
    
        time.sleep(1)  # 先让上面的任务向Queue存入数据，然后再让下面的任务开始从中取数据
    
        po.apply_async(reader, (q,))
        po.close()
        po.join()
        print("(%s) End" % os.getpid())
    '''
    (3112) start
    writer启动(5812),父进程为(3112)
    reader启动(15792),父进程为(3112)
    reader从Queue获取到消息：i
    reader从Queue获取到消息：t
    reader从Queue获取到消息：c
    reader从Queue获取到消息：a
    reader从Queue获取到消息：s
    reader从Queue获取到消息：t
    (3112) End
    
    '''

**统一声明：关于原创博客内容，可能会有部分内容参考自互联网，如有原创链接会声明引用；如找不到原创链接，在此声明如有侵权请联系删除哈。关于转载博客，如有原创链接会声明；如找不到原创链接，在此声明如有侵权请联系删除哈。**