Python多进程编程及多进程间的通信,数据传输

4,720 阅读12分钟
原文链接: www.cnblogs.com

多进程编程及进程间的通信多进程的优缺点 进程(process)三态 五态(三态的基础上增加了新建态和终止态)进程优先级 进程特征 孤儿进程 僵尸进程要求理解 多进程编程进程相关的函数 多进程模块Process() 创建自定义继承Process类 进程池技术 Pool函数进程间的通讯(IPC) 管道通讯多进程管道传输数据示例 消息队列单进程示例 多进程消息队列传递数据 共享内存 信号通信

 

多进程编程及进程间的通信

  • 意义:充分利用计算机的资源提高程序的运算速率
  • 定义:通过应用程序利用计算机多个核心达到同时执行多个任务的目的,以此提高计算机的运行速率
  • 实施方案:多进程 多线程
  • 并行: 计算机同时处理多个任务
  • 并发:同时处理多个任务,内核在不断的任务间小虎切换,达到好像还都在处理运行的效果,但是实际是一个时间点内核只能处理其中一个任务

多进程的优缺点

  • 优点

    • 可以使用计算机多核,进行任务的并发执行,提高执行效率
    • 空间独立,数据安全
    • 运行不受其他进程影响,创建方便
  • 缺点

    • 进程的删除和创建消耗的系统资源多

进程(process)

查看进程:ps -aux 查看进程树:pstree 查看父进程:ps -ajx

标志 名称 说明
S 等待态 可中断等待
D 等待态 不可中断等待
T 等待态 暂停状态
R 运行态 包含就绪状态
Z 僵尸进程  
<   高优先级
N   优先级较低
l   有子进程的
s   会话组组长
+   前台进程

三态

  • 就绪态:进程具备执行条件,等待系统分配资源
  • 运行态:进程占有CPU,处于运行状态
  • 等待态:进程暂时不具备执行条件,阻塞等待,满足条件后再执行

五态(三态的基础上增加了新建态和终止态)

  • 新建态:创建一个新的进程,获取资源的过程
  • 终止态:进程执行结束,资源释放回收的过程

进程优先级

  • 作用:决定了一个进程的执行权限

  • 动态查看系统中的进程信息:top,用< , >翻页

    • 取值范围:-20 -- 19 -20优先级最高
  • 使用指定的优先级运行程序

    • nice : 指定运行的优先级

      e.g. :nice -9 ./while.py --->>以优先级-9运行

进程特征

  1. 进程之间运行互不影响,各自独立运行
  2. 进程是操作系统资源分配的最小单位
  3. 每个进程空间独立,各自占有一定的虚拟内存

孤儿进程

  1. 父进程先于子进程退出,此时子进程就称为孤儿进程
  2. 孤儿进程会被操作系统指定的进程收养,系统进程就成为了孤儿进程的父进程

僵尸进程

  1. 子进程先于父进程退出,但是父进程没有处理子进程的退出状态,此时子进程就会成为僵尸进程
  2. 僵尸进程会存留少量的PCB信息在内存中,大量的僵尸进程会消耗系统资源,应该避免僵尸进程的产生
  • 如何避免僵尸进程的产生

    • 处理子进程的退出状态

      pid, status = os.wait()

       

      功能:在父进程中堵塞等待子进程退出

      返回值:

要求理解

  1. 什么是进程?
  2. 了解进程特征
  3. 清楚进程每种状态,

多进程编程

 
 1  
 2   
 3 # 功能 : 创建新的进程
 4 # 参数 : 无
 5 # 返回值 : 失败--->返回一个负数
 6 #  成功--->在新的进程中返回 ‘0’
 7 # 在原有的进程中返回新的进程的PID号
 8 import os
 9 pid = os.fork()
10 pid = os.fork()
11 print(pid)
12 ​
13 '''打印
14 9352
15 0
16 9353
17 '''

 

  • 子进程会复制父进程的全部代码,包括fork之前产生的内存空间
  • 子进程从fork的下一句开始执行,与父进程互不干扰
  • 父子进程的执行顺序不一定,父子进程公用一个终端显示
  • 父子进程通常会根据fork返回值的差异选择执行不同的代码。所以if结构几乎是fork的标配
  • 父子进程空间独立,操作都是本空间的内容,互不影响
  • 子进程也有自己的特性,比如PID号,PCB,命令集等。

进程相关的函数

函数方法 参数 说明
os.getpid()   返回当前进程的PID号
os.getppid()   返回当前进程的父进程的PID号
os._exit( status ) 程序的退出状态 进程退出
sys.exit( [ status ] ) 数字:表示退出状态,不写默认为 进程退出
  
import os
pid = os.fork()
if pid < 0:
    print("创建进程失败")
elif pid == 0:
    print("子进程我的真实PID为:",os.getpid(),"我的父进程PID为:",os.getppid())
else:
    print("我是父进程执行的代码,当前的变量pid为:",pid,"我的真实PID为:",os.getpid())
    
'''打印内容''''''
我是父进程执行的代码,当前的变量pid为: 10992 我的真实PID为: 10991
子进程我的真实PID为: 10992 我的父进程PID为: 10991
''''''打印内容'''
    
# 如果pid 10992和子进程真实PID不同,那么这个子进程就变成了孤儿进程

 

多进程模块

import multiprocessing

from multiprocessing import Process

Process()

  • 功能:创建一个进程对象

  • 参数

    • name:进程名称
    • target:绑定函数
    • args:元组,给target函数按照位置传参
    • kwargs:字典,给target函数按照键值对传参
    • name: 字符串,新的进程的名字
    • 例如:p = Process(target = fun,args=(a,b))
函数方法 说明
p.start() 启动进程,target函数自动执行,此时进程被真正创建
p.join([timeout]) 阻塞等待回收子进程,timeout为超时时间
p.is_alive() 判断进程生命周期状态,处于生命周期,返回布尔值
p.name() 获取进程名称
p.pid() 获取进程 的pid
p.daemon() 默认状态False,主进程退出不影响子进程。True :子进程随着主进程结束
  • 使用multiprocessing创建子进程,同样子进程复制父进程的全部代码,父子进程各自执行互不影响,父子进程有各自的运行空间

  • 如果不使用join挥手子进程则子进程退出后会成为僵尸进程

from multiprocessing import Process 
from time import sleep 
​
#带参数的进程函数
def worker(sec,name):
    for i in range(3):
        sleep(sec)
        print("I'm %s"%name)
        print("I'm working...")
​
p = Process(target = worker,args = (2,),\
    kwargs = {'name':'Daivl'},name = "Worker")
p.start()
​
print("Process name:",p.name) #进程名称
print("Process PID:",p.pid) #获取进程PID
​
#进程alive情况
print("Process is alive:",p.is_alive())
​
p.join(3)
print("==================")

 

 

 

创建自定义继承Process类

  1. 继承Process

  2. 编写自己的__init__,同时加载父类的__init__方法

  3. 重写run方法,可以通过生成的对象调用start自动运行

from  import Process
import time
​
class ClockProcess(Process):
​
    def __init__(self,value):
        self.value = value
        super(ClockProcess,self).__init__()
​
    def run(self):
        for i in range(5):
            print("现在的时间是",time.ctime())
            time.sleep(self.value)
​
# 创建自定义进的类的对象
p =ClockProcess(2)
​
# 自动调用run
p.start()
p.join()    

 

进程池技术

  • 产生原因

    • 如果有大量任务需要多进程完成。则需要频繁的创建删除进程,给计算机带来较多的资源消耗
  • 原理

    • 创建适当的进程放入进程池,用来池里待处理的时间,处理完当前任务后进程不销毁,仍然哎进程池等待处理其他时间,进程的复用降低了系统资源的消耗
  • 使用方法

    1. 创建进程池,在池内放入适当的进程
    2. 将事件加入到事件等待队列
    3. 不断取进程执行时间,直到所有进程执行完毕
    4. 关闭进程池,回收进程

Pool函数

  • Pool(Processes)

    • 功能:创建进程池对象
    • 表示进程池中有多少进程
  • pool.apply_async(func, args, kwds)

    • 功能:将时间放入进程池队列

    • 参数

      • func:事件函数
      • args:以元组形式给func传参
      • kwds : 以字典形式给func传参
    • 返回值一个对象

  • pool.apply(func, args, kwds)

    • 功能:将时间放入进程池队列

    • 参数

      • func:事件函数
      • args:以元组形式给func传参
      • kwds : 以字典形式给func传参
  • pool.close()

    • 功能:关闭进程池
  • pool.join()

    • 功能:回收进程池
  • pool.map(func,iter)

    • 功能:将要做的事件放入进程池

    • 参数

      • func : 要执行的函数
      • iter : 迭代对象
    • 返回值:多次return的结果列

from multiprocessing import Process
import time
​
class ClockProcess(Process):
​
    def __init__(self,value):
        self.value = value
        super(ClockProcess,self).__init__()
​
    def run(self):
        for i in range(5):
            print("现在的时间是",time.ctime())
            time.sleep(self.value)
​
# 创建自定义进的类的对象
p =ClockProcess(2)
​
# 自动调用run
p.start()
p.join()    

 

进程间的通讯(IPC)

进程间通讯的方法有:管道,消息队列,共享内存,信号,信号量,套接字

  管道 消息队列 共享内存
开辟空间 内存 内存 内存
读写方式 两端读写[双向/单向] 先进先出 覆盖之前的内容
效率 一般 一般 较高
应用 多用于父子进程 广泛灵活 需要注意互斥

 

管道通讯

通信原理:在内存中开辟管道空间,生成管道操作对象,多个进程使用“同一个”管道对象进行操作,即可实现通信

multiprocessing --> Pipe

  • fd1,fd2 = Pipe(duplex = True)

    • 功能:创建管道

    • 参数

      • 默认表示为双向管道
      • 设置False则为单向管道
    • 返回值

      • 如果是双向管道则都可以读写
      • 如果是单向管道,则fd1只读,fd2只写
  • fd.recv()

    • 功能:从管道读取信息
    • 返回值:读取到的内容
    • 如果管道为空则堵塞
  • fd.send(data)

    • 功能:向管道中写入内容
    • 参数:要写入的内容
    • 可以发送任意Python数据类型

多进程管道传输数据示例 

from multiprocessing import Process,Pipe
import time,os
​
# 创建管道对象
fd1, fd2 = Pipe()
​
def fun(name):
    time.sleep(1)
    fd2.send(os.getppid())
​
jobs = []
# 创建5个子进程
for i in range(5):
    p = Process(target = fun,args = (i,))
    jobs.append(p)
    p.start()
​
for i in range(5):
    data = fd1.recv()
    print(data)
​
for i in jobs:
    i.join()

 

 

消息队列

队列:先进先出,按照顺序来

通信原理:在内存中建立队列数据结构模型。多个进程都可以通过队列存入内容,取出内容的顺序和存入的顺序保持一致

  • 创建队列

    q = Queue(maxsize = 0)

    • 功能:创建队列消息
    • 参数:表示最多存储多少消息。默认表示根据内存分配存储
    • 返回值:队列对象
  • q.put(data, [block, timeout])

    • 功能:像队列存储消息

    • 参数

      • data:要存的内容
      • block:默认队列满时候会堵塞,设置False则非堵塞
      • timeout:超时时间
  • data = q.get([block, timeout])

    • 功能:获取队列消息

    • 参数

      • block:默认设置队列空时会堵塞,设置为False则非堵塞
      • timeout:超时时间
    • 返回值:返回取出的内容

  • q.full():判断队列是否为满

  • q.empty():判断队列是否为空

  • q.size():判断队列中的消息数量

  • q.close:关闭队列

单进程示例

#!《 单进程 》
from multiprocessing import Queue
from time import sleep
​
# 创建队列,可以放3条消息
q = Queue(3)
# 存一条消息
q.put(1)
sleep(0.5)
# 判断队列是否为空
print(q.empty())
q.put(2)
# 判断队列是否满
print(q.full())
q.put(3)
# 输出队列消息数量
print(q.qsize())
# 输出
print(q.get())
q.close()

 

 

多进程消息队列传递数据

from multiprocessing import Queue,Process
from time import sleep
​
# 创建队列,可以放3条消息
q = Queue(3)
​
def fun1():
    sleep(1)
    q.put({"a":1,"b":2})
​
def fun2():
    sleep(2)
    print("收到消息",q.get())
​
p1 = Process(target = fun1)
p2 = Process(target = fun2)
p1.start()
p2.start()
p1.join()
p2.join()

 

 

 

共享内存

通信原理:在内存中开辟一段空间,对多个进程可见,进程可以写入可以读,但是每次写入的内容会覆盖上一次的内容。

只能进行单个数据的发送

  • obj = Value(ctype, obj)

    • 功能:开辟共享内存空间

    • 参数

      • ctype:要存储的数据类型
      • obj:共享内存的初始数据
    • 返回值:共享内存对象

    • obj.value即共享内存值,对其修改即可修改内存

    • from multiprocessing import Value
      from time import sleep
      import os
      # 创建共享内存对象
      money = Value('i',2000)
      ​
      # 操作共享内存
      def deposite():
          while True:
              i = int(input("请输入:"))
              money.value = i
              sleep(0.05)
      ​
      def withdraw():
          data = money.value
          while True:
              if data != money.value :
                  data = money.value
                  print(data)
                  
      pid = os.fork()
      if pid == 0 :
          deposite()
      withdraw() 
  • obj = Array(ctype, obj)

    • 功能:开辟共享内存空间

    • 参数:

      • ctype:要存入的数据格式
      • obj:初始化存入的内容,比如列表、字符串。如果是数字则代表开辟内存空间的个数
    • 返回值:返回共享内存对象,可以遍历 

    • from multiprocessing import Array,Process
      from time import sleep
      import os
      ​
      # 开辟100字符内存空间,'c'代表字符,'i'代表整形
      shm = Array('c',100)
      # 必须使用字节流
      shm.value = "哈哈哈".encode()
      def fun1():
          print(os.getpid(),"子进程1:",shm.value.decode())
          shm.value = "夜夜夜".encode()
      ​
      def fun2():
          sleep(1)
          print(os.getpid(), "子进程2:",shm.value.decode())
      ​
      p1 = Process(target = fun1)  
      p2 = Process(target = fun2)
      p1.start()
      p2.start()
      p1.join()
      p2.join()
      ​
      ​

       

信号通信

一个进程向另一个进程发送一个信号来传递某种信息,接受者根据传递的信息来做相应的事

$ kill -l查看系统信号说明

$ kill -9 pid号对进程发送信号

信号名称 说明    
1) SIGHUP 连接断开    
2) SIGINT ctrl+c    
3) SIGQUIT ctrl+\    
20) SIGTSTP ctrl+z    
9) SIGKILL 终止进程    
19) SIGSTOP 暂停进程    
26) SIGVTALRM 时钟信号    
17) SIGCHLD 子进程退出时给父进程发的信号    
       

在Python中import signal可以获取信号

  • os.kill(pid, sig)

    • 功能:发送信号

    • 参数

      • pid:要发送信号的PID号
      • sig :信号名称 
import os
import signal
os.kill(12345,signal.SIGKILL) #杀死进程
作者:Banl
出处:www.cnblogs.com/BanL/ ----------------------------------------------------------------------

个性签名:苟日新 日日新 又日新!

如果这篇文章对你有些帮助,记得在右下角点个“推荐”,拜谢!