Python自学成才之路 进程间通信
程序在创建子进程时,会完全复制一份主进程的环境,包括变量,函数,类等。所以在子进程中使用的变量,函数,类和主进程之间隔离的,子进程之间也是隔离的。 看下面这个案例:
from multiprocessing import Process
AGE = 1
def hello():
print('hello')
def greet(names):
global AGE
AGE += 1
names.append('jack')
print('======子进程代码=======')
print('AGE = %s, id = %s'%(AGE, id(AGE)))
print('names = %s , id = %s'%(names, id(names)))
print('hello id = %s'%(id(hello)))
print('======子进程代码=======')
if __name__ == '__main__':
names = ['peter']
p = Process(target=greet, args=(names,))
p.start()
p.join()
print('第二次执行子进程')
p2 = Process(target=greet, args=(names,))
p2.start()
p2.join()
print('======父进程代码=======')
print('AGE = %s, id = %s'%(AGE, id(AGE)))
print('names = %s , id = %s'%(names, id(names)))
print('hello id = %s' % (id(hello)))
print('======父进程代码=======')
输出:
======子进程代码=======
AGE = 2, id = 140736619595456
names = ['peter', 'jack'] , id = 2672604171264
hello id = 2672609140448
======子进程代码=======
第二次执行子进程
======子进程代码=======
AGE = 2, id = 140736619595456
names = ['peter', 'jack'] , id = 1613780831232
hello id = 1613785865952
======子进程代码=======
======父进程代码=======
AGE = 1, id = 140736619595424
names = ['peter'] , id = 2006331253248
hello id = 2006329121232
======父进程代码=======
从案例中可以看到,进程把所有的变量,函数都新复制了一份,即使是全局变量进程之间也是隔离的。所以进程之间要想共享数据需要使用进程间的通信,进程间通行有两种方式,第一种是管道(Pipe),第二种是队列。
Pipe
- pipe常用于两个进程,两个进程分别位于管道的两端
- Pipe方法返回管道的两端conn1和conn2,Pipe有一个duplex参数,默认为True,即全双工模式,若为false,conn1只能接收消息,conn2只能发送消息。
class MyProcess(Process):
def __init__(self, threadname, conn):
super().__init__()
self.threadname = threadname
self.conn = conn
def run(self):
self.conn.send('我是' + self.threadname)
print(self.threadname + '-接受消息:' + self.conn.recv())
self.conn.close()
if __name__ == '__main__':
# 建立管道,拿到管道的两端,双工通信方式,两端都可以收发消息
conn1, conn2 = Pipe()
myProcess1 = MyProcess('thread1', conn1)
myProcess2 = MyProcess('thread2', conn2)
myProcess1.start()
myProcess2.start()
输出:
thread2-接受消息:我是thread1
thread1-接受消息:我是thread2
Queue Queue的一些常用方法的: Queue(n):初始化一个消息队列,并指定这个队列中最多能够容纳多少条消息。 put(obj,[block[,timeout]]):推入一条消息到这个队列中。默认是阻塞的,也就是说如果这个消息队列中已经满了,那么会会一直等待,将这个消息添加到消息队列中。timeout可以指定这个阻塞最长的时间,如果超过这个时间还是满的,就会抛出异常。 put_nowait() :非阻塞的推入一条消息,如果这个队列已经满了,那么会立马抛出异常。 qsize():获取这个消息队列消息的数量。 full():判断这个消息队列是否满了。 empty():判断这个消息队列是否空了。 get([block[,timeout]]):获取队列中的一条消息,然后将其从队列中移除,block默认为True。如果设置block为False,那么如果没值,会立马抛出异常。timeout指定如果多久没有获取到值后会抛出异常。
看个栗子:
from multiprocessing import Queue
# 初始化一个Queue对象,最多只能存放三条消息
q = Queue(3)
# 存放第一条消息
q.put('m1')
# 存放第二条消息
q.put('m2')
# 判断这个队列中是否已经满了
print(q.full())
# 存放第三条消息
q.put('m3')
# 判断这个队列中的消息是否已经满了
print(q.full())
# 因为如果消息队列已经满了,那么再put进去的时候就会报错
try:
# q.put('m4', block=False)# 队列满了立即抛出异常
# q.put('m4',block=True, timeout=2) # 可以被阻塞,等待时间超过两秒抛出异常
q.put_nowait('m4')
except:
print('消息队列已经满了,现有消息数量:%s' % q.qsize())
print(q.get())
输出:
False
True
消息队列已经满了,现有消息数量:3
m1
利用queue作为进程间通信来实现生产者和消费者
from multiprocessing import Process, Queue
import os
def write(q):
for x in ['m1', 'm2', 'm3']:
q.put(x)
print('子进程已经存放了消息%s, id : %s' % (x, os.getpid()))
def read(q):
while True:
try:
msg = q.get(block=True, timeout=3)
print('子进程已经读出了消息%s, id : %s' % (msg, os.getpid()))
except:
print('所有消息已经读完了')
break
if __name__ == '__main__':
q = Queue()
pw = Process(target=write, args=(q,))
pr = Process(target=read, args=(q,))
pw.start()
pr.start()
pw.join()
输出:
子进程已经存放了消息m1, id : 28740
子进程已经存放了消息m2, id : 28740
子进程已经读出了消息m1, id : 17936
子进程已经读出了消息m2, id : 17936
所有消息已经读完了
进程池间通信 进程池间的通信使用Manager().Queue(),不能使用Queue(会报错,Queue objects should only be shared between processes through inheritance),Manager().Queue()和Queue的使用方法是一样的。
from multiprocessing import Pool, Manager
import os
def write(q):
for x in ['m1', 'm2']:
q.put(x)
print('子进程已经存放了消息%s, id : %s' % (x, os.getpid()))
def read(q):
while True:
try:
msg = q.get(block=True, timeout=3)
print('子进程已经读出了消息%s, id : %s' % (msg, os.getpid()))
except:
print('所有消息已经读完了')
break
if __name__ == '__main__':
q = Manager().Queue()
pool = Pool(2)
pool.apply(write, args=(q,))
pool.apply(read, args=(q,))
pool.close()
pool.join()
输出:
子进程已经存放了消息m1, id : 3052
子进程已经存放了消息m2, id : 3052
子进程已经读出了消息m1, id : 22556
子进程已经读出了消息m2, id : 22556
所有消息已经读完了
做一下小总结:Python进程间数据是不共享的,所有的函数,变量,类都会被重新复制一份,要想让进程间可以共享数据,需要用到进程通信技术。比如pipe,和queue。Pipe常用于两个进程间的两端通信,实际用得较少。通常都是通过queue来实现进程间通信,进程池间通行是通过Manager.Queue。
- 细说Android事件传递
- swift基础_ set get方法 理解
- 高仿今日头条(2)
- ios tableview 上加 textfiled
- Spring Cloud实战小贴士:Feign的继承特性(伪RPC模式)
- 仿淘宝购买详情页购买缩小动画
- 高仿今日头条(1)
- android Native堆
- Spring Boot中Web应用的统一异常处理
- android bitmap的内存分配和优化
- Gitflow工作流程
- Android学习第一弹:Android通过用代码画虚线椭圆边框背景来学习一下shape的用法
- Spring Boot中使用Swagger2构建强大的RESTful API文档
- iOS iOS与html进行交互
- JavaScript 教程
- JavaScript 编辑工具
- JavaScript 与HTML
- JavaScript 与Java
- JavaScript 数据结构
- JavaScript 基本数据类型
- JavaScript 特殊数据类型
- JavaScript 运算符
- JavaScript typeof 运算符
- JavaScript 表达式
- JavaScript 类型转换
- JavaScript 基本语法
- JavaScript 注释
- Javascript 基本处理流程
- Javascript 选择结构
- Javascript if 语句
- Javascript if 语句的嵌套
- Javascript switch 语句
- Javascript 循环结构
- Javascript 循环结构实例
- Javascript 跳转语句
- Javascript 控制语句总结
- Javascript 函数介绍
- Javascript 函数的定义
- Javascript 函数调用
- Javascript 几种特殊的函数
- JavaScript 内置函数简介
- Javascript eval() 函数
- Javascript isFinite() 函数
- Javascript isNaN() 函数
- parseInt() 与 parseFloat()
- escape() 与 unescape()
- Javascript 字符串介绍
- Javascript length属性
- javascript 字符串函数
- Javascript 日期对象简介
- Javascript 日期对象用途
- Date 对象属性和方法
- Javascript 数组是什么
- Javascript 创建数组
- Javascript 数组赋值与取值
- Javascript 数组属性和方法
- 【Vim】vim配置的选择
- 关于HashMap的一些理解
- SpringBoot2.x系列教程(六十一)Spring Boot集成FastDFS实现文件上传
- 解决对接华为kafka 进行Kerberos认证时的sun.security.krb5.KrbException
- Newbe.Claptrap 框架如何实现多级生命周期控制?
- Kafka linux(centos7)安装配置kerberos并在另一台机器上使用Java服务接收消息
- SpringCloud bus 手动post可用 使用webhooks自动更新配置失效的解决小记
- 处理Storm1.2.2 日志报错 与spring boot冲突
- Java 中的 ==, equals 与 hashCode 的区别与联系
- 专家精品|密码朋克的社会实验(三):比特币发明了什么
- 面试题答案
- Vue3 为何使用 Proxy 实现数据监听
- vivo 悟空活动中台 - H5 活动加载优化
- PostgreSQL常用命令
- 实用!一键生成数据库文档,堪称数据库界的Swagger