个人中心
gpt4 book ai didi

Python multiprocessing多进程原理与应用示例

转载 作者：qq735679552 更新时间：2022-09-28 22:32:09 25 4
gpt4 key购买 nike

CFSDN坚持开源创造价值，我们致力于搭建一个资源共享平台，让每一个IT人在这里找到属于你的精彩世界.

这篇CFSDN的博客文章Python multiprocessing多进程原理与应用示例由作者收集整理，如果你对这篇文章有兴趣，记得点赞哟.

本文实例讲述了Python multiprocessing多进程原理与应用。分享给大家供大家参考，具体如下:

multiprocessing包是Python中的多进程管理包,可以利用multiprocessing.Process对象来创建进程,Process对象拥有is_alive()、join([timeout])、run()、start()、terminate()等方法.

multprocessing模块的核心就是使管理进程像管理线程一样方便，每个进程有自己独立的GIL，所以不存在进程间争抢GIL的问题，在多核CPU环境中，可以大大提高运行效率.

multiprocessing使用示例:

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
import multiprocessing
import time
import cv2
def daemon1(image):
   name = multiprocessing.current_process().name
   for i in range ( 50 ):
     image = cv2.GaussianBlur(image, ( 3 , 3 ), 1 )
     time.sleep( 0.1 )
   print 'daemon1 done!'
   cv2.imshow( 'daemon1' , image)
def daemon2(image):
   name = multiprocessing.current_process().name
   for i in range ( 50 ):
     image = cv2.GaussianBlur(image, ( 3 , 3 ), 1 )
     time.sleep( 0.5 )
   print 'daemon2 done!'
   cv2.imshow( 'daemon2' , image)
if __name__ = = '__main__' :
   t1 = time.time()
   number_kernel = multiprocessing.cpu_count()
   print 'We have {0} kernels' . format (number_kernel)
   p1 = multiprocessing.Process(name = 'daemon1' ,
                 target = daemon1,args = (cv2.imread( './p1.jpg' ),))
   p1.daemon = False
   p2 = multiprocessing.Process(name = 'daemon2' ,
                 target = daemon2, args = (cv2.imread( './p2.jpg' ),))
   p2.daemon = False
   p1.start()
   p2.start()
   print 'p1 is {0}' . format (p1.is_alive())
   p1.terminate()
   p1.join()
   print 'p1 is {0}' . format (p1.is_alive())
   print 'p2 is {0}' . format (p2.is_alive())
   p2.join()
   t2 = time.time()
   print '!!!!!!!!!!!!!!!!!!!!OK!!!!!!!!!!!!!!!!!!!!!'
   print 'total time is {0}' . format (t2 - t1)
   print 'p1.exitcode = {0}' . format (p1.exitcode)
   print 'p2.exitcode = {0}' . format (p2.exitcode)

multiprocessing中Process是一个类，用于创建进程，以及定义进程的方法，Process类的构造函数是:

?
1
def __init__( self , group = None , target = None , name = None , args = (), kwargs = {})

参数含义:

  • group：进程所属组，基本不用
  • target： 创建进程关联的对象，需要传入要多进程处理的函数名
  • name： 定义进程的名称
  • args： 表示与target相关联的函数的传入参数，可以传入多个，注意args是一个元组，如果传入的参数只有一个，需要表示为 args = (element1,)
  • kwargs： 表示调用对象的字典

程序解读:

  • multiprocessing.cpu_count()： 返回机器上cpu核的总数量
  • p1.daemon = False ： 定义子进程的运行属性，如果 .daemon设置为False表示子进程可以在主进程完成之后继续执行； 如果 .daemon设置为True，表示子进程随着主进程的结束而结束；必须在start之前设置；
  • p1.start(): 开始执行子进程p1
  • p1.join(): 定义子进程p1的运行阻塞主进程，只有p1子进程执行完成之后才会继续执行join之后的主进程，但是子进程间互相不受join影响。
  • 可以定义子进程阻塞主进程的时间--p1.join(100)，超时之后，主进程不再等待，开始执行。join()需要放在start()方法之后；
  • p1.terminate()：终止子进程的执行，其后要跟上jion()方法更新子进程的状态；
  • p1.exitcode： 进程的退出状态：  == 0 未生成任何错误，正常退出；  > 0 进程有一个错误，并以该错误码退出；    <0 进程由一个-1 * exitcode信号结束

在multiprocessing中使用pool 。

如果需要多个子进程时，使用进程池(pool)来（自动）管理各个子进程更加方便:

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from multiprocessing import Pool
import os, time
def long_time_task(name):
   print 'Run task {0} ({1})' . format (name,os.getpid())
   start = time.time()
   time.sleep( 3 )
   end = time.time()
   print 'Task {0} runs {1:.2f} seconds.' . format (name,end - start)
if __name__ = = '__main__' :
   print 'Parent process ({0})' . format (os.getpid)
   p = Pool()
   for i in range ( 12 ):
     p.apply_async(long_time_task, args = (i,))
   print 'Waiting for all subprocesses done...'
   p.close()
   p.join()
   print 'All subprocesses done.'

与Process类创建进程的方法不同，Pool是通过apply_async(func,args=(args))方法创建进程，一个进程池中能同时运行的任务数是机器上CPU核的总数量n_kernel，如果创建的子进程数大于n_kernel，则同时执行n_kernel个进程，这n_kernel中某个进程完成之后才会启动下一个进程.

  • os.getpid()是获取当前执行的进程的ID
  • p.close()方法是关掉进程池，表示不能再继续向进程池添加进程了。
  • p.join()方法是子进程阻塞主进程，必须在调用p.close()关闭进程池之后才能调用join()方法

多个子进程间的通信 。

多个子进程间的通信要用到multiprocessing.Queue，Queue的特性是它是一个消息队列。比如有以下的需求，一个子进程向队列中写数据，另外一个进程从队列中取数据的例子:

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from multiprocessing import Process, Queue
import os, time, random
def write(q):
   for value in [ 'A' , 'B' , 'C' ]:
     print 'Put {0} to queue...' . format (value)
     q.put(value)
     time.sleep(random.random())
def read(q):
   while True :
     if not q.empty():
       value = q.get( True )
       print 'Get {0} from queue.' . format (value)
       time.sleep(random.random())
     else :
       break
if __name__ = = '__main__' :
   q = multiprocessing.Queue()
   pw = Process(target = write, args = (q,))
   pr = Process(target = read, args = (q,))
   pw.start()
   pw.join()
   pr.start()
   pr.join()

Queue使用方法:

  • Queue.qsize()：返回当前队列包含的消息数量；
  • Queue.empty()：如果队列为空，返回True，反之False ；
  • Queue.full()：如果队列满了，返回True,反之False；
  • Queue.get():获取队列中的一条消息，然后将其从列队中移除，可传参超时时长；
  • Queue.get_nowait()：相当Queue.get(False),取不到值时触发异常：Empty；
  • Queue.put():将一个值添加进数列，可传参超时时长；
  • Queue.put_nowait():相当于Queue.get(False),当队列满了时报错：Full；

在进程池Pool中，使用Queue会出错，需要使用Manager.Queue:

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
from multiprocessing import Process, Queue
import os, time, random
def write(q):
   for value in [ 'A' , 'B' , 'C' ]:
     print 'Put {0} to queue...' . format (value)
     q.put(value)
     time.sleep(random.random())
def read(q):
   while True :
     if not q.empty():
       value = q.get( True )
       print 'Get {0} from queue.' . format (value)
       time.sleep(random.random())
     else :
       break
if __name__ = = '__main__' :
   manager = multiprocessing.Manager()
   q = manager.Queue()
   p = Pool()
   pw = p.apply_async(write, args = (q,))
   time.sleep( 2 )
   pr = p.apply_async(read, args = (q,))
   p.close()
   p.join()
   if not q.empty():
     print 'q is not empty...'
   else :
     print 'q is empty...'
   print 'OK'
   if not q.empty():
     print 'q is not empty...'
   else :
     print 'q is empty...'
   print 'done...'

父进程与子进程共享内存 。

定义普通的变量，不能实现在父进程和子进程之间共享:

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
import multiprocessing
from multiprocessing import Pool
def changevalue(n, a):
   n = 3.14
   a[ 0 ] = 5
if __name__ = = '__main__' :
   num = 0
   arr = range ( 10 )
   p = Pool()
   p1 = p.apply_async(changevalue, args = (num, arr))
   p.close()
   p.join()
   print num
   print arr[:]

结果输出num的值还是在父进程中定义的0，arr的第一个元素值还是0.

使用multiprocessing创建共享对象

?
1
2
3
4
5
6
7
8
9
10
11
12
import multiprocessing
def changevalue(n, a):
   n.value = 3.14
   a[ 0 ] = 5
if __name__ = = '__main__' :
   num = multiprocessing.Value( 'd' , 0.0 )
   arr = multiprocessing.Array( 'i' , range ( 10 ))
   p = multiprocessing.Process(target = changevalue, args = (num, arr))
   p.start()
   p.join()
   print num.value
   print arr[:]

结果输出num的值是在子进程中修改的3.14，arr的第一个元素值更改为5.

共享内存在Pool中的使用:

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
import multiprocessing
from multiprocessing import Pool
def changevalue(n, a):
   n.value = 3.14
   a[ 0 ] = 5
if __name__ = = '__main__' :
   num = multiprocessing.Value( 'd' , 0.0 )
   arr = multiprocessing.Array( 'i' , range ( 10 ))
   p = Pool()
   p1 = p.apply_async(changevalue, args = (num, arr))
   p.close()
   p.join()
   print num.value
   print arr[:]

希望本文所述对大家Python程序设计有所帮助.

原文链接：https://blog.csdn.net/dcrmg/article/details/78231845 。

最后此篇关于Python multiprocessing多进程原理与应用示例的文章就讲到这里了,如果你想了解更多关于Python multiprocessing多进程原理与应用示例的内容请搜索CFSDN的文章或继续浏览相关文章，希望大家以后支持我的博客！ 。

25 4 0
文章推荐： Tomcat如何修正JDK原生线程池Bug？
文章推荐： 使用SpringBoot请求参数过滤空格
文章推荐： 如何每小时改变你的 Linux 桌面壁纸
文章推荐： 一文理解Go 中的可寻址和不可寻址
qq735679552
个人简介

我是一名优秀的程序员,十分优秀！

滴滴打车优惠券免费领取
滴滴打车优惠券
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com