本文共 7524 字，大约阅读时间需要 25 分钟。

1、如何创建一个进程池

Pool类描述了一个工作进程池，他有几种不同的方法让任务卸载工作进程。

进程池内部维护一个进程序列，当使用时，则去进程池中获取一个进程，如果进程池序列中没有可供使用的进进程，那么程序就会等待，直到进程池中有可用进程为止。

我们可以用Pool类创建一个进程池， 展开提交的任务给进程池。 例：

#apply  (非阻塞，异步方式)from  multiprocessing import Poolimport timedef f1(i):    time.sleep(0.5)    print(i)    return i + 100if __name__ == "__main__":    pool = Pool(5)    for i in range(1,31):        pool.apply(func=f1,args=(i,))#apply_async   (阻塞，同步方式)def f1(i):    time.sleep(0.5)    print(i)    return i + 100def f2(arg):    print(arg)if __name__ == "__main__":    pool = Pool(5)    for i in range(1,31):        pool.apply_async(func=f1,args=(i,),callback=f2)    pool.close()    pool.join()

2、什么时候需要使用进程池

在利用Python进行系统管理的时候，特别是同时操作多个文件目录，或者远程控制多台主机，并行操作可以节约大量的时间。当被操作对象数目不大时，可以直接利用multiprocessing中的Process动态成生多个进程，十几个还好，但如果是上百个，上千个目标，手动的去限制进程数量却又太过繁琐，此时可以发挥进程池的功效。

Pool可以提供指定数量的进程供用户调用，当有新的请求提交到pool中时，如果池还没有满，那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到规定最大值，那么该请求就会等待，直到池中有进程结束，才会创建新的进程来它。

3、方法说明：

• apply_async(func[, args[, kwds[, callback]]]) 它是非阻塞，apply(func[, args[, kwds]])是阻塞的（理解区别，看4和5的结果区别）
• close()    关闭pool，使其不在接受新的任务。
• terminate()    结束工作进程，不在处理未完成的任务。
• join()    主进程阻塞，等待子进程的退出， join方法要在close或terminate之后使用。

• map()

  函数原型：map(func, iterable[, chunksize=None])

  Pool类中的map方法，与内置的map函数用法行为基本一致，它会使进程阻塞直到结果返回

  注意：虽然第二个参数是一个迭代器，但在实际使用中，必须在整个队列都就绪后，程序才会运行子进程

• map_async()

  函数原型：map_async(func, iterable[, chunksize[, callback]])

  与map用法一致，但是它是非阻塞的

4、非阻塞、异步方式使用进程池

举例：

#coding: utf-8import multiprocessingimport timedef func(msg):    print "msg:", msg    time.sleep(3)    print "end"if __name__ == "__main__":    pool = multiprocessing.Pool(processes = 3)    for i in xrange(4):        msg = "hello %d" %(i)        pool.apply_async(func, (msg, ))   #维持执行的进程总数为processes，当一个进程执行完毕后会添加新的进程进去    print "Mark~ Mark~ Mark~~~~~~~~~~~~~~~~~~~~~~"    pool.close()    pool.join()   #调用join之前，先调用close函数，否则会出错。执行完close后不会有新的进程加入到pool,join函数等待所有子进程结束    print "Sub-process(es) done."#结果：Mark~ Mark~ Mark~~~~~~~~~~~~~~~~~~~~~~msg: hello 0msg: hello 1msg: hello 2endmsg: hello 3endendendSub-process(es) done.

执行说明：创建一个进程池pool，并设定进程的数量为3，xrange(4)会相继产生四个对象[0, 1, 2, 4]，四个对象被提交到pool中，因pool指定进程数为3，所以0、1、2会直接送到进程中执行，当其中一个执行完事后才空出一个进程处理对象3，所以会出现输出“msg: hello 3”出现在"end"后。因为为非阻塞，主函数会自己执行自个的，不搭理进程的执行，所以运行完for循环后直接输出“mMsg: hark~ Mark~ Mark~~~~~~~~~~~~~~~~~~~~~~”，主程序在pool.join（）处等待各个进程的结束。

5、阻塞、同步方式使用线程池

举例：

#coding: utf-8import multiprocessingimport timedef func(msg):    print "msg:", msg    time.sleep(3)    print "end"if __name__ == "__main__":    pool = multiprocessing.Pool(processes = 3)    for i in xrange(4):        msg = "hello %d" %(i)        pool.apply(func, (msg, ))   #维持执行的进程总数为processes，当一个进程执行完毕后会添加新的进程进去    print "Mark~ Mark~ Mark~~~~~~~~~~~~~~~~~~~~~~"    pool.close()    pool.join()   #调用join之前，先调用close函数，否则会出错。执行完close后不会有新的进程加入到pool,join函数等待所有子进程结束    print "Sub-process(es) done."# 执行结果：msg: hello 0endmsg: hello 1endmsg: hello 2endmsg: hello 3endMark~ Mark~ Mark~~~~~~~~~~~~~~~~~~~~~~Sub-process(es) done.

可以看出apply方法和串行没有区别，只能等到前面一个进程执行完之后才能执行下一个进程。

6、apply和async_apply的区别

参考：

因为进程的切换是操作系统来控制的，抢占式的切换模式。

我们首先运行的是主进程，cpu运行很快啊，这短短的几行代码，完全没有给操作系统进程切换的机会，主进程就运行完毕了，整个程序结束。子进程完全没有机会切换到程序就已经结束了。

apply是阻塞式的。

首先主进程开始运行，碰到子进程，操作系统切换到子进程，等待子进程运行结束后，在切换到另外一个子进程，直到所有子进程运行完毕。然后在切换到主进程，运行剩余的部分。

apply_async是异步非阻塞式的。

首先主进程开始运行，碰到子进程后，主进程说：让我先运行个够，等到操作系统进行进程切换的时候，在交给子进程运行。以为我们的程序太短，然而还没等到操作系统进行进程切换，主进程就运行完毕了。

想要子进程执行，就告诉主进程：你等着所有子进程执行完毕后，在运行剩余部分。

python官方建议：废弃apply，使用apply_async。

7、使用线程池并关注结果(就是获取各个进程的返回值)

参考：

举例：

import multiprocessingimport timedef func(msg):    print('hello :', msg, time.ctime())    time.sleep(2)    print('end', time.ctime())    return 'done' + msgif __name__ == '__main__':    pool = multiprocessing.Pool(2)    result = []    for i in range(3):        msg = 'hello %s' % i        result.append(pool.apply_async(func=func, args=(msg,)))    pool.close()    pool.join()    for res in result:        print('***:', res.get())             # get()函数得出每个返回结果的值    print('All end--')

运行结果：

8、使用进程池执行多个不同的函数

举例：

import multiprocessingimport timeimport osdef Lee():    print('\nRun task Lee--%s******ppid:%s' % (os.getpid(), os.getppid()), '~~~~', time.ctime())    start = time.time()    time.sleep(5)    end = time.time()    print('Task Lee,runs %0.2f seconds.' % (end - start), '~~~~', time.ctime())def Marlon():    print("\nRun task Marlon-%s******ppid:%s" % (os.getpid(), os.getppid()), '~~~~', time.ctime())    start = time.time()    time.sleep(10)    end = time.time()    print('Task Marlon runs %0.2f seconds.' % (end - start), '~~~~', time.ctime())def Allen():    print("\nRun task Allen-%s******ppid:%s" % (os.getpid(), os.getppid()), '~~~~', time.ctime())    start = time.time()    time.sleep(15)    end = time.time()    print('Task Allen runs %0.2f seconds.' % (end - start), '~~~~', time.ctime())def Frank():    print("\nRun task Frank-%s******ppid:%s" % (os.getpid(), os.getppid()), '~~~~', time.ctime())    start = time.time()    time.sleep(20)    end = time.time()    print('Task Frank runs %0.2f seconds.' % (end - start), '~~~~', time.ctime())if __name__ == '__main__':    func_list = [Lee, Marlon, Allen, Frank]    print('parent process id %s' % os.getpid())    pool = multiprocessing.Pool(4)    for func in func_list:        pool.apply_async(func)    print('Waiting for all subprocesses done...')    pool.close()    pool.join()    print('All subprocesses done.')

结果：

parent process id 84172Waiting for all subprocesses done...Run task Lee--84868******ppid:84172 ~~~~ Thu Nov 14 17:44:14 2019Run task Marlon-84252******ppid:84172 ~~~~ Thu Nov 14 17:44:14 2019Run task Allen-85344******ppid:84172 ~~~~ Thu Nov 14 17:44:14 2019Run task Frank-85116******ppid:84172 ~~~~ Thu Nov 14 17:44:14 2019Task Lee,runs 5.00 seconds. ~~~~ Thu Nov 14 17:44:19 2019Task Marlon runs 10.00 seconds. ~~~~ Thu Nov 14 17:44:24 2019Task Allen runs 15.00 seconds. ~~~~ Thu Nov 14 17:44:29 2019Task Frank runs 20.00 seconds. ~~~~ Thu Nov 14 17:44:34 2019All subprocesses done.

9、使用async_apply非阻塞异步方式时调用回调函数callback：

from multiprocessing import Poolimport timedef fun_01(i):    time.sleep(2)    print('start_time:', time.ctime())    return i + 100def fun_02(arg):    print('end_time:', arg, time.ctime())if __name__ == '__main__':    pool = Pool(3)    for i in range(4):        pool.apply_async(func=fun_01, args=(i,), callback=fun_02)  # fun_02的入参为fun_01的返回值        # pool.apply_async(func=fun_01, args=(i,))    pool.close()    pool.join()    print('done')

运行结果：

# 立马一下子输出下面6行start_time: Thu Nov 14 16:31:41 2019end_time: 100 Thu Nov 14 16:31:41 2019start_time: Thu Nov 14 16:31:41 2019end_time: 101 Thu Nov 14 16:31:41 2019start_time: Thu Nov 14 16:31:41 2019end_time: 102 Thu Nov 14 16:31:41 2019# 暂停下后输出以下内容：start_time: Thu Nov 14 16:31:43 2019end_time: 103 Thu Nov 14 16:31:43 2019done

10、使用map（）和map_async

注意和直接使用apply_async的区别（参数列表的区别）

import timefrom multiprocessing import Pooldef run(fn):    # fn: 函数参数是数据列表的一个元素    time.sleep(1)    print(fn * fn)if __name__ == "__main__":    testFL = [1, 2, 3, 4, 5, 6]    print('shunxu:')  # 顺序执行(也就是串行执行，单进程)    s = time.time()    for fn in testFL:        run(fn)    t1 = time.time()    print("顺序执行时间：", int(t1 - s))    print('concurrent:')  # 创建多个进程，并行执行    pool = Pool(3)  # 创建拥有3个进程数量的进程池    # testFL:要处理的数据列表，run：处理testFL列表中数据的函数    pool.map(run, testFL)    pool.close()  # 关闭进程池，不再接受新的进程    pool.join()  # 主进程阻塞等待子进程的退出    t2 = time.time()    print("并行执行时间：", int(t2 - t1))

输出：

apply、apply_async与map及map_async的区别：

map_async()和map()方法，第2个参数可以是列表也可以是元祖，如下图：

而使用apply()和apply_async()方法时，第2个参数只能传入元祖，传入列表进程不会被执行，如下图：

11、python获取计算机的cpu数量

转载地址：https://blog.csdn.net/h_jlwg6688/article/details/108614426 如侵犯您的版权，请留言回复原文章的地址，我们会给您删除此文章，给您带来不便请您谅解！