欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

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

程序员文章站 2023-11-04 12:45:40
本文实例讲述了python multiprocessing多进程原理与应用。分享给大家供大家参考,具体如下: multiprocessing包是python中的多进程管理...

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

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

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

multiprocessing使用示例:

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类的构造函数是:

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)来(自动)管理各个子进程更加方便:

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的特性是它是一个消息队列。比如有以下的需求,一个子进程向队列中写数据,另外一个进程从队列中取数据的例子:

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

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...'

父进程与子进程共享内存

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

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创建共享对象:

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中的使用:

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相关内容感兴趣的读者可查看本站专题:《python进程与线程操作技巧总结》、《python数据结构与算法教程》、《python函数使用技巧总结》、《python字符串操作技巧汇总》、《python入门与进阶经典教程》、《python+mysql数据库程序设计入门教程》及《python常见数据库操作技巧汇总

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