文档章节

python中的Queue与多进程(multiprocessing)

Kevin_Yang
 Kevin_Yang
发布于 2014/07/30 00:53
字数 1574
阅读 42250
收藏 22
点赞 9
评论 3

最近接触一个项目,要在多个虚拟机中运行任务,参考别人之前项目的代码,采用了多进程来处理,于是上网查了查python中的多进程

一、先说说Queue(队列对象)

Queue是python中的标准库,可以直接import 引用,之前学习的时候有听过著名的“先吃先拉”与“后吃先吐”,其实就是这里说的队列,队列的构造的时候可以定义它的容量,别吃撑了,吃多了,就会报错,构造的时候不写或者写个小于1的数则表示无限多

import Queue

q = Queue.Queue(10)

向队列中放值(put)

q.put(‘yang’)

q.put(4)

q.put([‘yan’,’xing’])

在队列中取值get()

默认的队列是先进先出的

>>> q.get()
'yang'
>>> q.get()
4
>>> q.get()
['yan', 'xing']
>>>

 

当一个队列为空的时候如果再用get取则会堵塞,所以取队列的时候一般是用到

get_nowait()方法,这种方法在向一个空队列取值的时候会抛一个Empty异常

所以更常用的方法是先判断一个队列是否为空,如果不为空则取值

队列中常用的方法

Queue.qsize() 返回队列的大小 
Queue.empty() 如果队列为空,返回True,反之False 
Queue.full() 如果队列满了,返回True,反之False
Queue.get([block[, timeout]]) 获取队列,timeout等待时间 
Queue.get_nowait() 相当Queue.get(False)
非阻塞 Queue.put(item) 写入队列,timeout等待时间 
Queue.put_nowait(item) 相当Queue.put(item, False)

 

二、multiprocessing中使用子进程概念

from multiprocessing import Process

可以通过Process来构造一个子进程

p = Process(target=fun,args=(args))

再通过p.start()来启动子进程

再通过p.join()方法来使得子进程运行结束后再执行父进程

from multiprocessing import Process
import os

# 子进程要执行的代码
def run_proc(name):
    print 'Run child process %s (%s)...' % (name, os.getpid())

if __name__=='__main__':
    print 'Parent process %s.' % os.getpid()
    p = Process(target=run_proc, args=('test',))
    print 'Process will start.'
    p.start()
    p.join()
    print 'Process end.'

image

三、在multiprocessing中使用pool

如果需要多个子进程时可以考虑使用进程池(pool)来管理

from multiprocessing import Pool

 

from multiprocessing import Pool
import os, time

def long_time_task(name):
    print 'Run task %s (%s)...' % (name, os.getpid())
    start = time.time()
    time.sleep(3)
    end = time.time()
    print 'Task %s runs %0.2f seconds.' % (name, (end - start))

if __name__=='__main__':
    print 'Parent process %s.' % os.getpid()
    p = Pool()
    for i in range(5):
        p.apply_async(long_time_task, args=(i,))
    print 'Waiting for all subprocesses done...'
    p.close()
    p.join()
    print 'All subprocesses done.'

pool创建子进程的方法与Process不同,是通过

p.apply_async(func,args=(args))实现,一个池子里能同时运行的任务是取决你电脑的cpu数量,如我的电脑现在是有4个cpu,那会子进程task0,task1,task2,task3可以同时启动,task4则在之前的一个某个进程结束后才开始

image

上面的程序运行后的结果其实是按照上图中1,2,3分开进行的,先打印1,3秒后打印2,再3秒后打印3

代码中的p.close()是关掉进程池子,是不再向里面添加进程了,对Pool对象调用join()方法会等待所有子进程执行完毕,调用join()之前必须先调用close(),调用close()之后就不能继续添加新的Process了。

当时也可以是实例pool的时候给它定义一个进程的多少

如果上面的代码中p=Pool(5)那么所有的子进程就可以同时进行

三、多个子进程间的通信

多个子进程间的通信就要采用第一步中说到的Queue,比如有以下的需求,一个子进程向队列中写数据,另外一个进程从队列中取数据,

#coding:gbk

from multiprocessing import Process, Queue
import os, time, random

# 写数据进程执行的代码:
def write(q):
    for value in ['A', 'B', 'C']:
        print 'Put %s to queue...' % value
        q.put(value)
        time.sleep(random.random())

# 读数据进程执行的代码:
def read(q):
    while True:
        if not q.empty():
            value = q.get(True)
            print 'Get %s from queue.' % value
            time.sleep(random.random())
        else:
            break

if __name__=='__main__':
    # 父进程创建Queue,并传给各个子进程:
    q = Queue()
    pw = Process(target=write, args=(q,))
    pr = Process(target=read, args=(q,))
    # 启动子进程pw,写入:
    pw.start()    
    # 等待pw结束:
    pw.join()
    # 启动子进程pr,读取:
    pr.start()
    pr.join()
    # pr进程里是死循环,无法等待其结束,只能强行终止:
    print
    print '所有数据都写入并且读完'

四、关于上面代码的几个有趣的问题

if __name__=='__main__':    
    # 父进程创建Queue,并传给各个子进程:
    q = Queue()
    p = Pool()
    pw = p.apply_async(write,args=(q,))    
    pr = p.apply_async(read,args=(q,))
    p.close()
    p.join()
    
    print
    print '所有数据都写入并且读完'

如果main函数写成上面的样本,本来我想要的是将会得到一个队列,将其作为参数传入进程池子里的每个子进程,但是却得到

RuntimeError: Queue objects should only be shared between processes through inheritance

的错误,查了下,大意是队列对象不能在父进程与子进程间通信,这个如果想要使用进程池中使用队列则要使用multiprocess的Manager类

if __name__=='__main__':
    manager = multiprocessing.Manager()
    # 父进程创建Queue,并传给各个子进程:
    q = manager.Queue()
    p = Pool()
    pw = p.apply_async(write,args=(q,))
    time.sleep(0.5)
    pr = p.apply_async(read,args=(q,))
    p.close()
    p.join()
    
    print
    print '所有数据都写入并且读完'

 

这样这个队列对象就可以在父进程与子进程间通信,不用池则不需要Manager,以后再扩展multiprocess中的Manager类吧

关于锁的应用,在不同程序间如果有同时对同一个队列操作的时候,为了避免错误,可以在某个函数操作队列的时候给它加把锁,这样在同一个时间内则只能有一个子进程对队列进行操作,锁也要在manager对象中的锁

#coding:gbk

from multiprocessing import Process,Queue,Pool
import multiprocessing
import os, time, random

# 写数据进程执行的代码:
def write(q,lock):
    lock.acquire() #加上锁
    for value in ['A', 'B', 'C']:
        print 'Put %s to queue...' % value        
        q.put(value)        
    lock.release() #释放锁  

# 读数据进程执行的代码:
def read(q):
    while True:
        if not q.empty():
            value = q.get(False)
            print 'Get %s from queue.' % value
            time.sleep(random.random())
        else:
            break

if __name__=='__main__':
    manager = multiprocessing.Manager()
    # 父进程创建Queue,并传给各个子进程:
    q = manager.Queue()
    lock = manager.Lock() #初始化一把锁
    p = Pool()
    pw = p.apply_async(write,args=(q,lock))    
    pr = p.apply_async(read,args=(q,))
    p.close()
    p.join()
    
    print
    print '所有数据都写入并且读完'

参考文章:

http://blog.csdn.net/yatere/article/details/6668006

http://www.liaoxuefeng.com/wiki/001374738125095c955c1e6d8bb493182103fac9270762a000/0013868323401155ceb3db1e2044f80b974b469eb06cb43000

© 著作权归作者所有

共有 人打赏支持
Kevin_Yang
粉丝 247
博文 34
码字总数 34570
作品 0
通州
加载中

评论(3)

j
jiangzx
不错,谢谢!
立雪听风
立雪听风

引用来自“吐槽的达达仔”的评论

q = manager.Queue()
Mulitprocess里面的Queue不是线程安全的吗?还用Lock,是不是多此一举了??
赞同: Queue 是FIFO。就像 流水线作业,que=manager.Queue()
工人A put-->[queue:流水线] get--> (工人B)
(1)工人A::que.put(“产品”) , 如果流水线满了,则会阻塞等待直到流水线有空间;
(2)工人B:que.get(“产品”): 相反,如果queue空了,会阻塞等待直到流水线有产品

假设,工人A 、B 分别是一个进程(线程也可以),虽然他们是异步的,因为分别执行了 que.put() 、que.get() 形成了 B 等待 A 的同步关系。
另外,lock,不是锁住某个资源(文件/对象),而是解析器指令。线程及进程 比喻为学生。班上只有一台电脑(lock),老师(解析器)把电脑给你玩,那其它人就等着。如果你一直玩,不交出电脑。那么其它永远等着。这就是死锁!! 所以,要 try 电脑给你 finally 你交出电脑
吐槽的达达仔
吐槽的达达仔
q = manager.Queue()
Mulitprocess里面的Queue不是线程安全的吗?还用Lock,是不是多此一举了??
python--多进程的用法详解实例

想让python实现多进程(multiprocessing),我们要先区分不同的操作系统的不同之处。 Linux操作系统下提供了一个fork()系统调用,普通函数调用一次返回一次,fork()调用一次返回两次,因为操作...

山有木兮有木兮 ⋅ 05/14 ⋅ 0

Python怎么利用多核cpu

原文链接http://www.cnblogs.com/stubborn412/p/4033651.html def dead_loop(): def dead_loop(): {void DeadLoop() { while (true); } } from threading import Thread lib = cdll.LoadLibr......

dby_freedom ⋅ 05/06 ⋅ 0

Python - multiprocessing运行和停止进程

基本用法 本文只介绍控制进程启停的操作, 因此只用到一个类 首先写一个保持运行的方法: 初始化一个实例, target为该实例运行时执行的方法. 通过调用类的方法启动一个进程: 要停止一个进程实例...

严北 ⋅ 05/22 ⋅ 0

Python collections.deque vs. Queue.Queue vs. multiprocessing.Queue

总体上来说,当需要在进程间通信的时候需要使用multiprocessing.Queue; 当在同一个进程当中,而需要多线程之间通信的时候,可以使用Queue.Queue;而至于collections.deque一般就是在同一个线...

geekpy ⋅ 02/12 ⋅ 0

Python标准库10 多进程初步 (multiprocessing包)

作者:Vamei 出处:http://www.cnblogs.com/vamei 欢迎转载,也请保留这段声明。谢谢! 我们已经见过了使用subprocess包来创建子进程,但这个包有两个很大的局限性:1) 我们总是让subproces...

osDaniel ⋅ 2014/09/21 ⋅ 0

python高性能编程第一章读书笔记

计算机底层组件分为三大基本部分:计算单元、存储单元以及两者之间的连接。 计算单元:具有将接收到的任意输入转换成输出的能力以及改变当前处理状态的能力。CPU是最常见的计算单元。它的主要...

ma412410029 ⋅ 05/28 ⋅ 0

python多线程多进程并发

python多线程 1.基本使用方法,配合Queue并发完成任务. As is known to all,因为GIL,python的多线程并不能多核并发.下面这个例子是多线程配合subprocess子进程,Queue.Queue队列进行的多线程i...

flyking ⋅ 2013/10/12 ⋅ 0

Python3学习笔记11-标准库之子进程、信号、多线程、进程信息

关于进程与线程的简单理解(以工厂举例:cup-》工厂,车间-》进程,线程-》工人),可以参考阮一峰的博文进程与线程的一个简单图文解释 一、子进程 (subprocess包) 这里的内容以Linux进程基础...

Corwien ⋅ 2016/06/11 ⋅ 0

关于python multiprocessing进程通信的pipe和queue方式

这两天温故了python 的multiprocessing多进程模块,看到的pipe和queue这两种ipc方式,啥事ipc? ipc就是进程间的通信模式,常用的一半是socke,rpc,pipe和消息队列等。 今个就再把pipe和que...

rfyiamcool ⋅ 2014/09/08 ⋅ 0

Python进程学习

线程及进程概念可自行学习 Unix/Linux操作系统提供了一个fork()系统调用,它非常特殊。普通的函数调用,调用一次,返回一次,但是fork()调用一次,返回两次,因为操作系统自动把当前进程(称...

粗粮面包 ⋅ 2017/12/10 ⋅ 0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

Day 17 vim简介与一般模式介绍

vim简介 vi和Vim的最大区别就是编辑一个文件时vi不会显示颜色,而Vim会显示颜色。显示颜色更便于用户编辑,凄然功能没有太大的区别 使用 yum install -y vim-enhanced 安装 vim的三种常用模式...

杉下 ⋅ 37分钟前 ⋅ 0

【每天一个JQuery特效】根据可见状态确定是否显示或隐藏元素(3)

效果图示: 主要代码: <!DOCTYPE html><html><head><meta charset="UTF-8"><title>根据可见状态确定 是否显示或隐藏元素</title><script src="js/jquery-3.3.1.min.js" ty......

Rhymo-Wu ⋅ 46分钟前 ⋅ 0

OSChina 周四乱弹 —— 初中我身体就已经垮了,不知道为什么

Osc乱弹歌单(2018)请戳(这里) 【今日歌曲】 @加油东溪少年 :下完这场雨 后弦 《下完这场雨》- 后弦 手机党少年们想听歌,请使劲儿戳(这里) @马丁的代码 :买了日本 日本果然赢了 翻了...

小小编辑 ⋅ 今天 ⋅ 10

浅谈springboot Web模式下的线程安全问题

我们在@RestController下,一般都是@AutoWired一些Service,由于这些Service都是单例,所以并不存在线程安全问题。 由于Controller本身是单例模式 (非线程安全的), 这意味着每个request过来,...

算法之名 ⋅ 今天 ⋅ 0

知乎Java数据结构

作者:匿名用户 链接:https://www.zhihu.com/question/35947829/answer/66113038 来源:知乎 著作权归作者所有。商业转载请联系作者获得授权,非商业转载请注明出处。 感觉知乎上嘲讽题主简...

颖伙虫 ⋅ 今天 ⋅ 0

Confluence 6 恢复一个站点有关使用站点导出为备份的说明

推荐使用生产备份策略。我们推荐你针对你的生产环境中使用的 Confluence 参考 Production Backup Strategy 页面中的内容进行备份和恢复(这个需要你备份你的数据库和 home 目录)。XML 导出备...

honeymose ⋅ 今天 ⋅ 0

JavaScript零基础入门——(九)JavaScript的函数

JavaScript零基础入门——(九)JavaScript的函数 欢迎回到我们的JavaScript零基础入门,上一节课我们了解了有关JS中数组的相关知识点,不知道大家有没有自己去敲一敲,消化一下?这一节课,...

JandenMa ⋅ 今天 ⋅ 0

火狐浏览器各版本下载及插件httprequest

各版本下载地址:http://ftp.mozilla.org/pub/mozilla.org//firefox/releases/ httprequest插件截至57版本可用

xiaoge2016 ⋅ 今天 ⋅ 0

Docker系列教程28-实战:使用Docker Compose运行ELK

原文:http://www.itmuch.com/docker/28-docker-compose-in-action-elk/,转载请说明出处。 ElasticSearch【存储】 Logtash【日志聚合器】 Kibana【界面】 答案: version: '2'services: ...

周立_ITMuch ⋅ 今天 ⋅ 0

使用快嘉sdkg极速搭建接口模拟系统

在具体项目研发过程中,一旦前后端双方约定好接口,前端和app同事就会希望后台同事可以尽快提供可供对接的接口方便调试,而对后台同事来说定好接口还仅是个开始、设计流程,实现业务逻辑,编...

fastjrun ⋅ 今天 ⋅ 0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

返回顶部
顶部