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
" % 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