进程简单使用
Process([group [, target [, name [, args [, kwargs]]]]])
- target:如果传递了函数的引用,可以任务这个子进程就执行这里的代码
- args:给target指定的函数传递的参数,以元组的方式传递
- kwargs:给target指定的函数传递命名参数
- name:给进程设定一个名字,可以不设定
- group:指定进程组,大多数情况下用不到
- Process创建的实例对象的常用方法:
start():启动子进程实例(创建子进程)
- is_alive():判断进程子进程是否还在活着
- join([timeout]):是否等待子进程执行结束,或等待多少秒
- terminate():不管任务是否完成,立即终止子进程
Process创建的实例对象的常用属性:
- name:当前进程的别名,默认为Process-N,N为从1开始递增的整数
- pid:当前进程的pid(进程号)
import multiprocessing import time def work1(): while True: time.sleep(1) print("------ in work1 ------") def work2(): while True: time.sleep(1) print("------ in work2 ------") if __name__ == '__main__': multiprocessing.Process(target=work1).start() multiprocessing.Process(target=work2).start()
进程间通信
初始化Queue()对象时(例如:q=Queue()),若括号中没有指定最大可接收的消息数量,或数量为负值,那么就代表可接受的消息数量没有上限(直到内存的尽头);
- Queue.qsize():返回当前队列包含的消息数量;
- Queue.empty():如果队列为空,返回True,反之False ;
- Queue.full():如果队列满了,返回True,反之False;
- Queue.get([block[, timeout]]):获取队列中的一条消息,然后将其从列队中移除,block默认值为True;
import multiprocessing def down_load(queue): data = [11, 22, 33, 44] for temp in data: queue.put(temp) print "=====正在下载数据=====" def analysis_data(queue): """ 数据处理 :return: """ while not queue.empty(): print queue.get() print "=====完成解析=====" def main(): # 创建一个队列 queue = multiprocessing.Queue() # 创建多个进程,将队列的应用作为实参进行传递 p1 = multiprocessing.Process(target=down_load, args=(queue, )) p2 = multiprocessing.Process(target=analysis_data, args=(queue, )) p1.start() p2.start() if __name__ == '__main__': main()
进程池
multiprocessing.Pool常用函数解析:
- apply_async(func[, args[, kwds]]) :使用非阻塞方式调用func(并行执行,堵塞方式必须等待上一个进程>- 退出才能执行下一个进程),args为传递给func的参数列表,kwds为传递给func的关键字参数列表;
- close():关闭Pool,使其不再接受新的任务;
- terminate():不管任务是否完成,立即终止;
- join():主进程阻塞,等待子进程的退出, 必须在close或terminate之后使用;
from multiprocessing import Pool import os, time, random def worker(msg): t_start = time.time() print "%s 开始执行, 进程号为:%d" % (msg, os.getpid()) time.sleep(random.random()*2) t_stop = time.time() print "%s 执行完毕,耗时%0.2f" % (msg, t_stop-t_start) if __name__ == '__main__': po = Pool(3) for i in range(10): # Pool().apply_async(要调用的目标,(传递给目标的参数元祖,)) # 每次循环将会用空闲出来的子进程去调用目标 po.apply_async(worker, (i,)) print "========Start========" po.close() # 关闭进程池,关闭后po不再接收新的请求 po.join() # 等待po中所有子进程执行完成,必须放在close语句之后 print "========End========="