python 归纳 (十五)_多进程使用Pool

Posted 2021-01-05 sunzebo

1 usePool.py

#coding: utf-8
"""
学习进程池使用 multiprocessing.Pool
总结:
  1. Pool 池用于处理 多进程，并不是多线程
  2. 池有大小的概念
  3. 并不是所有的子进程添加完了，才开始启动子进程。 当第一个进程添加到池中的时候，马上就启动了

使用:
  1. 创建进程池对象 pool = multiprocessing.Pool(processes = 3)
  2. 往池中添加进程  主要:pool.apply_async(func, (参数, )) or pool.apply(func, (参数, ))
  3. 调用 pool.close(); pool.join() (一般伴随 apply_async)，等待所有子进程结束

其他：
terminate()    结束工作进程，不再处理未完成的任务
map(...) 将一个集合数据 映射到 同一个函数，  根据集合大小 执行多次子进程
get()  从子进程获取返回结果
"""
import multiprocessing
import time

# 进程代码
def func(msg):
    print "sub begin:", msg
    time.sleep(2)
    print "sub end:",msg

if __name__ == "__main__":
    pool = multiprocessing.Pool(processes = 3) # 创建进程池
    for i in xrange(5):
        msg = " %d" %(i)

        #  apply_async 非阻塞,一般和join一起使用, apply 阻塞 主进程等待子进程一个接一个执行完
        #  apply_async 维持执行的进程总数为processes，当一个进程执行完毕后会添加新的进程进去
        #  apply_async 这里创建的都是守护进程
        pool.apply_async(func, (msg, ))  # 实际开发中，每个子线程执行不同的逻辑
        time.sleep(1)
        print "alread start sub,%d\\n" % i

    print "Mark~ Mark~ Mark~~~~~~~~~~~~~~~~~~~~~~"
    pool.close() # 关闭pool使其不在接受新的任务，必须有
    pool.join()  # 等待所有子进程结束  调用join之前，先调用close函数，
    print "Sub-process(es) done."

"""
pool.apply_async
Out:

sub begin:  0
alread start sub,0

sub begin:  1
alread start sub,1

sub begin:  2
sub end:  0
sub end:  1alread start sub,2


sub begin:  3
alread start sub,3

sub begin:  4
sub end:  2
sub end:alread start sub,4
 3

Mark~ Mark~ Mark~~~~~~~~~~~~~~~~~~~~~~
sub end:  4
Sub-process(es) done.

"""

"""
pool.apply
Out:

sub begin:  0
sub end:  0
alread start sub,0

sub begin:  1
sub end:  1
alread start sub,1

sub begin:  2
sub end:  2
alread start sub,2

sub begin:  3
sub end:  3
alread start sub,3

sub begin:  4
sub end:  4
alread start sub,4

Mark~ Mark~ Mark~~~~~~~~~~~~~~~~~~~~~~
Sub-process(es) done.

"""

 

2 usePoolmap.py

# -*- coding: utf-8 -*-
"""
使用 multiprocessing.Pool.map 执行多进程

逻辑：
   有10个大小的列表，进程池4个大小
   使用map执行完

总结：
   可以简化启动子进程代码

使用:
    1. 创建进程池对象 pool = multiprocessing.Pool(processes = 3)
    2. 准备list 数据  i_list = range(10)
    3. 准备子进程执行代码 函数 sub_process_code
    4. 调用 pool.map(sub_process_code, i_list)
       或
            pool.map_async(sub_process_code, i_list)
            pool.close()
            pool.join()

"""
import multiprocessing
import time
import os
def sub_process_code(x):
    # 打印  hh:ss 编号 进程ID
    print time.strftime(\'%M:%S\',time.localtime(time.time())),x * x,os.getpid()
    time.sleep(3)

if __name__ == \'__main__\':
    pool = multiprocessing.Pool(multiprocessing.cpu_count()) # 根据CPU数量创建进程池，这里是4个
    i_list = range(10)
    pool.map(sub_process_code, i_list)

    ## 下面3行代码 = 上面一行代码
    # pool.map_async(sub_process_code, i_list) # 异步
    # pool.close()
    # pool.join()  # 如果没有join，主进程 结束后，所有子进程马上结束了
    print "end"

"""
Out:

24:20 0 5960
24:20 1 5840
24:20 4 5892
24:20 9 6944
24:23 16 5960
24:23 25 5840
24:23 36 5892
24:23 49 6944
24:26 64 5960
24:26 81 5840
end

"""

 

3 usePoolgetData.py

# -*- coding: utf-8 -*-
"""
使用进程池 multiprocessing.Pool，获取子进程的返回数据

使用:
    1. 创建进程池对象 pool = multiprocessing.Pool(processes = 3)
    2. 往池中添加进程，同时拿到Result对象 p_ApplyResult_obj = pool.apply_async(func, (参数, ))
    3. 调用 pool.close(); pool.join() 等待所有子进程结束
    4. 获取子进程的返回数据  p_ApplyResult_obj.get()
"""

import multiprocessing
import time

#  子进程代码，会return 数据给主进程
def func(msg):
    time.sleep(3)
    print "end"
    return "return " + msg

if __name__ == "__main__":
    pool = multiprocessing.Pool(processes=4)
    result = []   # 存储Result对象

    for i in xrange(3):
        msg = "hello %d" %(i)
        # 添加子进程的同时，获取它的返回对象
        p_ApplyResult_obj = pool.apply_async(func, (msg, ))
        print id(p_ApplyResult_obj)  # 打印pool对象 ID
        result.append(p_ApplyResult_obj)
    pool.close()
    pool.join()
    for res in result:
        print ":::", res.get()   # 获取子进程的return结果
    print "Sub-process(es) done."

"""
Out:

41974752
41974864
41975032
endend

end
::: return hello 0
::: return hello 1
::: return hello 2
Sub-process(es) done.
"""

 

4 usePoolCallback.py 使用回调函数  

# -*- coding: utf-8 -*-
"""
进程池  回调函数

逻辑: 子进程返回值结束，主进程马上调用回调 函数接收返回值，并打印 
总结:
    1、回调函数是子进程结束时，由主进程调用的函数
    2、个人理解  类似 事件-驱动-动作机制  信号-动作机制  数据库触发器机制
        事件 子进程执行结束
        动作 执行回调函数

用法:
    1. 准备子进程函数Foo， 里面有返回值
    2. 准备回调函数Bar，接收参数就是子进程返回值
    3. pool.apply_async(func=Foo, args=(i,), callback=Bar)
      Foo的参数 是 args
      Bar的参数 是 Foo的返回值

"""
from multiprocessing import  Pool
import time

def Foo(i):
    """
    字进程代码
    :param i:
    """
    time.sleep(2)
    print i
    return  i+100

def Bar(arg):
    """
    回调函数
    :param arg:    子进程执行代码返回值 ，本例是 Foo  return i+100
    """
    print(\'----->exec done:\',arg)


if __name__ == \'__main__\':

    # 允许进程池里同时放入5个进程
    pool = Pool(5)

    #  启动10个进程
    for i in range(10):

        # 并行执行，callback回调执行者为父进程
        pool.apply_async(func=Foo, args=(i,), callback=Bar)

        #pool.apply(func=Foo, args=(i,))  # 串行执行

    print(\'end\')
    pool.close()
    pool.join()  # 进程池中进程执行完毕后再关闭，如果注释，那么程序直接关闭。

"""
Out:

end
0
(\'----->exec done:\', 100)
1
(\'----->exec done:\', 101)
2
(\'----->exec done:\', 102)
3
(\'----->exec done:\', 103)
4
(\'----->exec done:\', 104)
5
(\'----->exec done:\', 105)
6
(\'----->exec done:\', 106)
7
(\'----->exec done:\', 107)
8
(\'----->exec done:\', 108)
9
(\'----->exec done:\', 109)
"""

 

参考：

 

python进程池：multiprocessing.pool