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

本文实例讲述了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程序设计有所帮助。

时间: 2019-02-25

Python多进程池 multiprocessing Pool用法示例

本文实例讲述了Python多进程池 multiprocessing Pool用法.分享给大家供大家参考,具体如下: 1. 背景 由于需要写python程序, 定时.大量发送htttp请求,并对结果进行处理. 参考其他代码有进程池,记录一下. 2. 多进程 vs 多线程 c++程序中,单个模块通常是单进程,会启动几十.上百个线程,充分发挥机器性能.(目前c++11有了std::thread编程多线程很方便,可以参考我之前的博客) shell脚本中,都是多进程后台执行.({ ...} &, 可以参考

简单学习Python多进程Multiprocessing

1.1 什么是 Multiprocessing 多线程在同一时间只能处理一个任务. 可把任务平均分配给每个核,而每个核具有自己的运算空间. 1.2 添加进程 Process 与线程类似,如下所示,但是该程序直接运行无结果,因为IDLE不支持多进程,在命令行终端运行才有结果显示 import multiprocessing as mp def job(a,b): print('abc') if __name__=='__main__': p1=mp.Process(target=job,args=

Python3多进程 multiprocessing 模块实例详解

本文实例讲述了Python3多进程 multiprocessing 模块.分享给大家供大家参考,具体如下: 多进程 Multiprocessing 模块 multiprocessing 模块官方说明文档 Process 类 Process 类用来描述一个进程对象.创建子进程的时候,只需要传入一个执行函数和函数的参数即可完成 Process 示例的创建. star() 方法启动进程, join() 方法实现进程间的同步,等待所有进程退出. close() 用来阻止多余的进程涌入进程池 Pool 造

Python标准库之多进程(multiprocessing包)介绍

在初步了解Python多进程之后,我们可以继续探索multiprocessing包中更加高级的工具.这些工具可以让我们更加便利地实现多进程. 进程池 进程池 (Process Pool)可以创建多个进程.这些进程就像是随时待命的士兵,准备执行任务(程序).一个进程池中可以容纳多个待命的士兵. "三个进程的进程池" 比如下面的程序: 复制代码 代码如下: import multiprocessing as mul def f(x):     return x**2 pool = mul.

Python多进程并发(multiprocessing)用法实例详解

本文实例讲述了Python多进程并发(multiprocessing)用法.分享给大家供大家参考.具体分析如下: 由于Python设计的限制(我说的是咱们常用的CPython).最多只能用满1个CPU核心. Python提供了非常好用的多进程包multiprocessing,你只需要定义一个函数,Python会替你完成其他所有事情.借助这个包,可以轻松完成从单进程到并发执行的转换. 1.新建单一进程 如果我们新建少量进程,可以如下: import multiprocessing import t

python基于multiprocessing的多进程创建方法

本文实例讲述了python基于multiprocessing的多进程创建方法.分享给大家供大家参考.具体如下: import multiprocessing import time def clock(interval): while True: print ("the time is %s"% time.time()) time.sleep(interval) if __name__=="__main__": p = multiprocessing.Process

Python多进程multiprocessing用法实例分析

本文实例讲述了Python多进程multiprocessing用法.分享给大家供大家参考,具体如下: mutilprocess简介 像线程一样管理进程,这个是mutilprocess的核心,他与threading很是相像,对多核CPU的利用率会比threading好的多. 简单的创建进程: import multiprocessing def worker(num): """thread worker function""" print 'Wor

Python多进程multiprocessing.Pool类详解

multiprocessing模块 multiprocessing包是Python中的多进程管理包.它与 threading.Thread类似,可以利用multiprocessing.Process对象来创建一个进程.该进程可以允许放在Python程序内部编写的函数中.该Process对象与Thread对象的用法相同,拥有is_alive().join([timeout]).run().start().terminate()等方法.属性有:authkey.daemon(要通过start()设置)

Python多进程库multiprocessing中进程池Pool类的使用详解

问题起因 最近要将一个文本分割成好几个topic,每个topic设计一个regressor,各regressor是相互独立的,最后汇总所有topic的regressor得到总得预测结果.没错!类似bagging ensemble!只是我没有抽样.文本不大,大概3000行,topic个数为8,于是我写了一个串行的程序,一个topic算完之后再算另一个topic.可是我在每个topic中用了GridSearchCV来调参,又要选特征又要调整regressor的参数,导致参数组合一共有1782种.我真

对python遍历文件夹中的所有jpg文件的实例详解

python发现文件夹下所有的jpg文件,并且安装文件排放的顺序输出 glob模块是最简单的模块之一,内容非常少.用它可以查找符合特定规则的文件路径名.跟使用windows下的文件搜索差不多.查找文件只用到三个匹配符:"*", "?", "[]"."*"匹配0个或多个字符:"?"匹配单个字符:"[]"匹配指定范围内的字符,如:[0-9]匹配数字. glob.glob 返回所有匹配的文件路

Java 中DateUtils日期工具类的实例详解

Java 中DateUtils日期工具类的实例详解 介绍 在java中队日期类型的处理并不方便,通常都需要借助java.text.SimpleDateFormat类来实现日期类型 和字符串类型之间的转换,但是在jdk1.8之后有所改善,jdk1.7以及之前的版本处理日期类型并不方便, 可以借助Joda Time组件来处理,尤其是日期类型的一些数学操作就更是不方便. java代码 /** * * 日期工具类 java对日期的操作一直都很不理想,直到jdk1.8之后才有了本质的改变. * 如果使用的

Android Studio 中aidl的自定义类的使用详解

自己折腾了好久,记录一下. service端: 1:创建类Dog,需要实现Parcelable接口: 2:aidl下创建 Dog.aidl,里面两句话就可以了 (1)package s包名; (2)parcelable Dog; 3:interface.aidl引入Dog类, import s包名.Dog; Client 端: 1:创建类Dog,需要实现Parcelable接口: 2:aidl下创建 Dog.aidl, (1)package c包名; (2)parcelable Dog; 注意:

Java中mybatis关于example类的使用详解

这几天刚接触example,很多内容都是破碎的,写一篇博文加深理解. 一.什么是example类 mybatis-generator会为每个字段产生如上的Criterion,如果表的字段比较多,产生的Example类会十分庞大.理论上通过example类可以构造你想到的任何筛选条件.在mybatis-generator中加以配置,配置数据表的生成操作就可以自动生成example了.具体配置可以参考MBG有关配置.      下面是mybatis自动生成example的使用. 二.了解exampl

Python编程对列表中字典元素进行排序的方法详解

本文实例讲述了Python编程对列表中字典元素进行排序的方法.分享给大家供大家参考,具体如下: 内容目录: 1. 问题起源 2. 对列表中的字典元素排序 3. 对json进行比较(忽略列表中字典的顺序) 一.问题起源 json对象a,b a = '{"ROAD": [{"id": 123}, {"name": "no1"}]}' b = '{"ROAD": [{"name": "

Python爬取网页中的图片(搜狗图片)详解

前言 最近几天,研究了一下一直很好奇的爬虫算法.这里写一下最近几天的点点心得.下面进入正文: 你可能需要的工作环境: Python 3.6官网下载 本地下载 我们这里以sogou作为爬取的对象. 首先我们进入搜狗图片http://pic.sogou.com/,进入壁纸分类(当然只是个例子Q_Q),因为如果需要爬取某网站资料,那么就要初步的了解它- 进去后就是这个啦,然后F12进入开发人员选项,笔者用的是Chrome. 右键图片>>检查 发现我们需要的图片src是在img标签下的,于是先试着用

numpy库与pandas库axis=0,axis= 1轴的用法详解

对数据进行操作时,经常需要在横轴方向或者数轴方向对数据进行操作,这时需要设定参数axis的值: axis = 0 代表对横轴操作,也就是第0轴: axis = 1 代表对纵轴操作,也就是第1轴: numpy库中横轴.纵轴 axis 参数实例详解: In [1]: import numpy as np #生成一个3行4列的数组 In [2]: a = np.arange(12).reshape(3,4) In [3]: a Out[3]: array([[ 0, 1, 2, 3], [ 4, 5,

Python 多进程并发操作中进程池Pool的实例

在利用Python进行系统管理的时候,特别是同时操作多个文件目录,或者远程控制多台主机,并行操作可以节约大量的时间.当被操作对象数目不大时,可以直接利用multiprocessing中的Process动态成生多个进程,10几个还好,但如果是上百个,上千个目标,手动的去限制进程数量却又太过繁琐,这时候进程池Pool发挥作用的时候就到了. Pool可以提供指定数量的进程,供用户调用,当有新的请求提交到pool中时,如果池还没有满,那么就会创建一个新的进程用来执行该请求:但如果池中的进程数已经达到规定

Python多进程multiprocessing、进程池用法实例分析

本文实例讲述了Python多进程multiprocessing.进程池用法.分享给大家供大家参考,具体如下: 内容相关: multiprocessing: 进程的创建与运行 进程常用相关函数 进程池: 为什么要有进程池 进程池的创建与运行:串行.并行 回调函数 多进程multiprocessing: python中的多进程需要使用multiprocessing模块 多进程的创建与运行: 1.进程的创建:进程对象=multiprocessing.Process(target=函数名,args=(参