39.创建多进程及进程通讯 -- Queue--Pipe--Event
2024-09-02 15:35:47
创建多进程
- windows:进程、线程
- linux:进程、线程(做了进程通信的多进程实现的线程)
- 进程之间内存彼此独立,不管是父子进程还是单个独立进程
- multiprocessing:Process 创建多进程python内置的模块
- current_process().name 返回的是当前的进程是哪个
from multiprocessing import process,current_process
def work():
print('我是进程:%s' % current_process().name)
work() 运行结果:我是进程:MainProcess- dir(current_process()) :current_process的内置函数
['_Popen', '__class__', '__delattr__', '__dict__', '__dir__', '__doc__', '__eq__', '__format__', '__ge__', '__getattribute__', '__gt__', '__hash__', '__init__', '__le__', '__lt__', '__module__', '__ne__', '__new__', '__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__sizeof__', '__str__', '__subclasshook__', '__weakref__', '_bootstrap', '_config', '_identity',
'_name', '_parent_pid', '_popen', 'authkey', 'daemon', 'exitcode', 'ident', 'is_alive', 'join', 'name', 'pid', 'run', 'sentinel', 'start', 'terminate']- current_process().pid:是打印当前进程id:返回的是当前进程的id
- current_process().name:返回的是当前的进程名称
- current_process().terminate:直接终止进程
- current_process().is_alive():返回进程的存活状态 True/False
- current_process().exitcode:0代表进程死亡 None代表进程运行
- current_process().ident:和pid类似
- 创建多进程其实就是把一个任务(工作函数)绑定在一个子进程上,然后将任务分配多个多个子进程上去执行
- 大家记得一定要把多进程的创建放在main函数里面
- help(process) :创建多进程函数帮助文档,target:调用函数名 ,name:子进程命名,args:元组传参,kwargs:字典传参
__init__(self, group=None, target=None, name=None, args=(), kwargs={}, *
, daemon=None)- p1 = Process(target=func,name=,args=(),kwargs={}):创建子进程传参模型
- p1.start():开启进程
- p1.join():回收子进程
- p1.name :#获取进程名字
- p1.pid :获取进程ID之
- p1.terminate :直接终止进程
- p1.is_alive() :返回进程的存活状态 True/False
- p1.exitcode: 0代表进程死亡 None代表进程运行
- p1.ident :和pid类似
- import sys sys.stdout.flush() #刷新缓存区
#创建多进程1.py
from multiprocessing import Process,current_process
import sys
def work():
print('我是子进程:%s进程id:%s' % (current_process().name,current_process().pid))
sys.stdout.flush() #刷新缓存
def main():
print('当前父进程:%s进程id:%s' % (current_process().name,current_process().pid))
p1 = Process(target=work,name='子进程一号') #创建子进程
p2 = Process(target=work,name='子进程二号') #创建子进程
p1.start() #开启进程
p2.start()
print(p1.name,p1.pid,p1.is_alive(),p1.exitcode,p1.ident)
print(p2.name,p2.pid,p2.is_alive(),p2.exitcode,p2.ident)
p1.join() #回收子进程
p2.join()
if __name__ == "__main__":
main()
##当前进程数是3个,一个父进程两个子进程运行结果:
当前父进程:MainProcess进程id:134216
子进程一号 129748 True None 129748
子进程二号 134780 True None 134780
我是子进程:子进程一号进程id:129748
我是子进程:子进程二号进程id:134780#创建多进程2.py
from multiprocessing import Process,current_process
import sys
def work(a,char):
print('-----------------------------')
print('当前子进程:%s' % current_process().name)
a[0] = char
print('当前子进程修改完毕:',a)
sys.stdout.flush() #刷新缓存
def main():
a = [1,2,3]
print('当前父进程:',a)
p1 = Process(target=work,name='进程1',args=(a,'a'))
p2 = Process(target=work,name='进程2',args=(a,'zhang'))
p1.start()
p2.start()
p1.join()
p2.join()
if __name__ == "__main__":
main()运行结果:
当前父进程: [1, 2, 3]
-----------------------------
当前子进程:进程1
当前子进程修改完毕: ['a', 2, 3]
-----------------------------
当前子进程:进程2
当前子进程修改完毕: ['zhang', 2, 3]
僵尸进程
- 如果父进程没有对子进程合理回收
- 父进程没有在子进程工作结束之后立即回收
- PCB控制块 PID继续占用 僵尸进程
- 子进程结束 ,父进程依然存活
from multiprocessing import Process,current_process
from time import sleep
def work():
print('子进程开始了')
print('PID:%s' % current_process().pid )
print('子进程结束了')
def main():
p = Process(target=work)
p.start() #父进程开启了子进程
#p.join() #回收子进程 或者忘记回收
#父进程退出了
sleep(10) #退出
#p.join()
if __name__ == '__main__':
main()
孤儿进程
- Python代码中如果在父进程没有明确写出join回收子进程,在父进程结束之后,父进程回自动帮助回收
- kill -9 pid 强制杀死
多进程通讯--Queue
- 》》》》Python中任何普通的基础数据类型,都不可以在多进程下通信
- **Queue**(size):共享通讯队列,阻塞的行为,默认为阻塞,block=False设置为非阻塞
- 值满了不能放:q.put(block=True) 默认为True
- 值空了不能取,会一直等:q.get(block=True)
- 当修改了拿取的方式为非阻塞,那么数据在取不到或者立即放不进去的时候会直接报错
- queue.Full 满了
- queue.Empty 空的
- q.empty():判断队列是否为空
- q.full():判断队列是否为满
- q.qsize():返回队列数据个数
#创建共享队列1.py
from multiprocessing import Process,current_process,Queue
def main():
q = Queue() #创建共享队列
print('当前父进程:',q)
for var in range(5):
q.put(var)
print(q.get())
print(q.get())
print(q.get())
print(q.get())
print(q.get())
if __name__ == "__main__":
main()运行结果:
当前父进程: <multiprocessing.queues.Queue object at 0x000000000277A2E8>
0
1
2
3
4#创建共享队列2.py
from multiprocessing import Process,current_process,Queue
def product(q):#生产数据
for var in range(5):
q.put(var,block=False)#非阻塞
def custom(q):#消费数据
for var in range(5):
i = q.get(block=False)#非阻塞
print('%s取到的数据:%s' % (current_process().name,i))
def main():
q = Queue(5) #创建共享队列
p1 = Process(target=product,name='生产进程',args=(q,))
p2 = Process(target=custom,name='消费进程',args=(q,))
p1.start(),p2.start(),p1.join(),p2.join()
if __name__ == "__main__":
main()运行结果:
消费进程取到的数据:0
消费进程取到的数据:1
消费进程取到的数据:2
消费进程取到的数据:3
消费进程取到的数据:4#创建共享队列3.py
from multiprocessing import Process,Queue,current_process
from time import sleep
import sys
def product(q):#某个进程生产数据
for var in range(10):
q.put(var)
sleep(1)
def custom(q):#这个进程消费数据
for var in range(5):
i = q.get()
print('%s进程取到的数据:%s' % (current_process().name,i))
sys.stdout.flush()
def main():
q = Queue() #共享队列
p1 = Process(target=product,args=(q,))
p2 = Process(target=custom,name='消费进程1',args=(q,))
p3 = Process(target=custom,name='消费进程2',args=(q,))
p1.start(),p2.start(),p3.start(),p1.join(),p2.join(),p3.join()
print('------------')
if __name__ == '__main__':
main()运行结果:
消费进程1进程取到的数据:0
消费进程1进程取到的数据:1
消费进程2进程取到的数据:2
消费进程1进程取到的数据:3
消费进程2进程取到的数据:4
消费进程1进程取到的数据:5
消费进程2进程取到的数据:6
消费进程1进程取到的数据:7
消费进程2进程取到的数据:8
消费进程2进程取到的数据:9
------------- 下面的是子进程while循环取值,在阻塞行为下值空了不能取,会一直等,那么如何才能不让它退出,也不报错呢?如果改为非阻塞那么值空了就会报错,看下面的方法
from multiprocessing import Process,Queue,current_process
from time import sleep
import sys
def product(q,sig):#某个进程生产数据
for var in range(8):
sig.put(True) #每次生产,都放一个True
q.put(var)
sleep(1)
else:
sig.put(False)
def custom(q,sig):#这个进程消费数据
while True:
if sig.get(): #代表还有数据要生产
i = q.get()
print('%s进程取到的数据:%s' % (current_process().name,i))
sys.stdout.flush()
else: #没有数据要生产了
print('子进程结束,数据生产完毕')
sig.put(False)
break
def main():
q = Queue() #共享队列
sig = Queue()
p1 = Process(target=product,args=(q,sig))
p2 = Process(target=custom,name='消费进程1',args=(q,sig))
p3 = Process(target=custom,name='消费进程2',args=(q,sig))
p1.start(),p2.start(),p3.start(),p1.join(),p3.join(),p2.join()
print('------------')
if __name__ == '__main__':
main()运行结果为:
消费进程1进程取到的数据:0
消费进程1进程取到的数据:1
消费进程2进程取到的数据:2
消费进程1进程取到的数据:3
消费进程2进程取到的数据:4
消费进程1进程取到的数据:5
消费进程2进程取到的数据:6
消费进程1进程取到的数据:7
消费进程2:子进程结束,数据生产完毕
消费进程1:子进程结束,数据生产完毕
------------
多进程通讯--pipe
- Pipe:管道两端, duplex=True 存储的都是pickle数据类型
- left,right = Pipe(duplex)
- pickle = p.recv() 向管道这一侧取出来
- p.send(pickle) 向管道另一侧去发
- 管道只能存放和拿取出来pickle数据类型
- pickle 二进制数据 Python中维持数据类型保存 解析
- pickle.loads:解析数据
- pickle.dumps:封装为二进制
- 管道在使用的时候,是有两端的,要注意使用顺序
- 首先:多个进程可以享用同一个端,也就是一个管道支持多个进程通信,非常安全的操作(要记住)
#pipe 单个消费者生产者模型
from multiprocessing import Process,current_process,Pipe
import pickle
def work_a(p):
for var in range(5):
p.send(pickle.dumps(var))
print('%s:生产数据|%s' % (current_process().name,var))
def work_b(p):
for var in range(5):
pickle_obj = pickle.loads(p.recv()) #阻塞
print('%s:消费数据|%s' % (current_process().name,pickle_obj))
def main():
a,b = Pipe() #创建管道
p1 = Process(target=work_a,name='生产者',kwargs={'p':a})
p2 = Process(target=work_b,name='消费者',kwargs={'p':b})
p1.start()
p2.start()
p1.join()
p2.join()
if __name__ == "__main__":
main()
- 运行结果:
生产者:生产数据|0
生产者:生产数据|1
生产者:生产数据|2
生产者:生产数据|3
生产者:生产数据|4
消费者:消费数据|0
消费者:消费数据|1
消费者:消费数据|2
消费者:消费数据|3
消费者:消费数据|4 #pipe 多个消费者生产者模型 一个消费者只消费一个数据,公司里面常用的模型,一个管道支持多个进程通信,非常安全的操作
from multiprocessing import Process,current_process,Pipe
import pickle
def work_a(p):
for var in range(5):
p.send(pickle.dumps(var))
print('%s:生产数据|%s' % (current_process().name,var))
def work_b(p):
pickle_obj = pickle.loads(p.recv()) #阻塞
print('%s:消费数据|%s' % (current_process().name,pickle_obj))
def main():
a,b = Pipe() #创建管道
p1 = Process(target=work_a,name='生产者',args=(a,))
p2 = [] #消费者进程队列
for var in range(5):
p2.append(Process(target=work_b,name='消费者%d' % var,args=(b,)))
p1.start() #下开启生产者
for var in p2:
var.start()
p1.join()
for var in p2:
var.join()
if __name__ == "__main__":
main()运行结果:
生产者:生产数据|0
生产者:生产数据|1
生产者:生产数据|2
生产者:生产数据|3
生产者:生产数据|4
消费者1:消费数据|0
消费者2:消费数据|1
消费者3:消费数据|2
消费者0:消费数据|3
消费者4:消费数据|4
多进程状态通讯--Event
- from multiprocessing import Event
- e = Event() e:状态实例
- e.set() #设置当前的实例状态为True
- e.clear() #设置当前的实例状态为False
- e.wait()
- False:阻塞等待
- True:向下执行
from multiprocessing import Event,Process,current_process
from time import sleep
import sys
def work(e):
print('%s已开启' % current_process().name)
sys.stdout.flush() #刷新输出缓冲区,立竿见影看到打印
e.wait() #Sleep 可中断睡眠
print('%s正式开始工作' % current_process().name)
sys.stdout.flush()
print('%s|pid:%s' % (current_process().name,current_process().pid))
def main():
print('主进程开启')
e = Event()
p1 = Process(target=work,name='进程1',args=(e,))
p2 = Process(target=work,name='进程2',args=(e,))
p1.start() #进程已经开启
p2.start()
for var in range(3):
print(var)
sleep(1) #- e.wait()
e.set() #信号变True
p1.join()
p2.join()
if __name__ == '__main__':
main()
运行结果:
主进程开启
0
进程2已开启
进程1已开启
1
2
进程1正式开始工作
进程2正式开始工作
进程1|pid:3108
进程2|pid:6068
最新文章
- linux系统ftp命令
- 卸载Centos自带open-jdk
- java基础之 重排序
- android XMl 解析神奇xstream 三: 把复杂对象转换成 xml
- Source Tree for MAC1.6
- 转载:SMS软件FESWMS模型使用体会
- java提供了native2ascii工具
- [Hive - LanguageManual] Hive Default Authorization - Legacy Mode
- spoj 178
- JavaScript的“闭包”到底是什么(2)
- sqlserver编程基本语法
- AspNetPager使用指南
- ngin隐藏版本号
- org.springframework.expression.spel.SpelEvaluationException: EL1011E: Method call: Attempted to call method test() on null context object
- 快速掌握Shell编程
- TreeNode.trage的使用
- Groovy 设计模式 -- 责任链模式
- Java_Number(装箱和拆箱)
- 课堂小练习(complex类)
- 正则 js分转元带千分符号