41.进程池--Pool

进程池

方便创建，管理进程，单独进程的Process创建，需要手动开启，维护任务函数，以及释放回收
进程池不需要这么麻烦，进程提前创建好，未来在使用的时候，可以直接给与任务函数
某个进程池中的任务结束了，占用的进程会自己释放刚才工作的事情，以便接收下一个
P = Pool(num) #创建一个包含有num个空闲进程的池子
p.apply() 填充任务，任务如果结束，会自动释放掉当前占用的进程
创建大规模任务，Pool(100)
1，创建进程池：进程池中的进程是可以复用的

- from mutliprocessing import Pool
- p = Pool(num)
  - num：指明当前多少空闲进程创建出来
- p.apply(func,args,)
  - 阻塞行为
  - func：指明填充功能函数名
  - args：对应的参数
  - 阻塞行为相当于(lock)加锁的进程池工作方式，有序的，第一个执行完才会执行第二个
- p.apply_async(func,args,)
  - 非阻塞行为，并发的，无序的
- p.close()
  - 在整个业务结束之后，进程池要首先关闭
  - 关闭之后进程池里的旧任务会继续执行但是没有办法填充新的任务
  - 进程池关闭了就无法打开
- p.join()
  - 进程回收，把关闭了的进程池中的每个进程join() 释放回收掉
- p.terminate()
  - 直接关闭进程池，并且终止所欲偶进程
2，进程池的工作的返回值：
- res = p.apply(func,)
  - res就是进程池的工作结果
  - 立竿见影就可以看到结果，就因为apply填充任务是阻塞行为
- res = p.apply_aysnc(func,)
  - 非阻塞的结果，可以立即拿到，但是不是结果，只是一个抽象虚拟的值
    - 这个值代表进程结束后的返回值
  - res.get() #使用要谨慎
    - 当前非阻塞执行的进程，有优先级先结束
    - 强制要求立即这个结果，但是会影响进程之间的并发效果
3，进程池中的通信队列是特殊的
- from multiprocessing import Manager
- q = Manager().Queue() 　　#进程共享队列
- 无法使用管道（Pipe）

#进程池创建
from multiprocessing import Pool
import sys
def work_a():
	for var in range(1,5):
		print(var) 
		sys.stdout.flush()
def work_b():
	for var in range(5,10):
		print(var)
		sys.stdout.flush()
def work_c():
	for var in range(10,15):
		print(var)
		sys.stdout.flush()
def main():
	p = Pool(2) #参数：可以最多同时执行任务个数，并不是填充的最大任务个数
	#p.apply_async(func=work,args=(a,b),) 非阻塞行为
	p.apply(func=work_a)#阻塞行为
	p.apply(func=work_b)
	p.apply(func=work_c)
	#?: 是否是阻塞行为执行完这三个任务
		#阻塞的话：1个等一个，同步
		#非阻塞：异步
	p.close()
	p.join()
if __name__ == '__main__':
	main()

运行结果：

#获取阻塞进程池返回结果
from multiprocessing import Pool
import sys
def work_a():
	for var in range(1,5):
		print(var) 
		sys.stdout.flush()
	return 'a'
def work_b():
	for var in range(5,10):
		print(var)
		sys.stdout.flush()
	return 'b'
def work_c():
	for var in range(10,15):
		print(var)
		sys.stdout.flush()
	return 'c'
def main():
	p = Pool(2) #参数：可以最多同时执行任务个数，并不是填充的最大任务个数
	res1 = p.apply(func=work_a) #阻塞的一个等一个，res1执行完才会执行res2
	res2 = p.apply(func=work_b)
	res3 = p.apply(func=work_c)
	print('res1进程返回结果:%s' % res1),print('res2进程返回结果:%s' % res2),print('res3进程返回结果:%s' % res3)
	p.close(),p.join()
if __name__ == '__main__':
	main()

运行结果：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
res1进程返回结果:a
res2进程返回结果:b
res3进程返回结果:c

#获取非阻塞进程池返回结果
from multiprocessing import Pool
import sys
def work_a():
	for var in range(1,5):
#		print(var) 
		sys.stdout.flush()
	return 'a'
def work_b():
	for var in range(5,10):
#		print(var)
		sys.stdout.flush()
	return 'b'
def work_c():
	for var in range(10,15):
#		print(var)
		sys.stdout.flush()
	return 'c'
def main():
	p = Pool(2) #参数：可以最多同时执行任务个数，并不是填充的最大任务个数
	res1 = p.apply_async(func=work_a) #非阻塞的会返回一个抽象的数据
	res2 = p.apply_async(func=work_b)
	res3 = p.apply_async(func=work_c)
	print('res1进程返回结果:%s' % res1.get()),print('res2进程返回结果:%s' % res2.get()),print('res3进程返回结果:%s' % res3.get())
	p.close(),p.join()
if __name__ == '__main__':
	main()

运行结果：

res1进程返回结果:a
res2进程返回结果:b
res3进程返回结果:c

#进程池通讯--Queue
from multiprocessing import Pool,Manager,Queue
from time import sleep
import sys
def work_a(q):
	#生产者 放十次
	for var in range(10):
		print('生产者:',var) 
		sys.stdout.flush()
		q.put(var)
		sleep(1)
def work_b(q):
	#消费者，拿十次
	for var in range(10):
		res = q.get() #阻塞行为
		print('消费者:',var)
		sys.stdout.flush()
def main():
	q = Manager().Queue() #进程共享队列
	p = Pool(5) #进程可以复用
	p.apply_async(func=work_a,args={q,q})
	p.apply_async(func=work_b,args={q,q})
	p.close()
	p.join()
if __name__ == '__main__':
	main()

运行结果：

生产者: 0
消费者: 0
生产者: 1
消费者: 1
生产者: 2
消费者: 2
生产者: 3
消费者: 3
生产者: 4
消费者: 4
生产者: 5
消费者: 5
生产者: 6
消费者: 6
生产者: 7
消费者: 7
生产者: 8
消费者: 8
生产者: 9
消费者: 9