Python教程

一文攻克Python的进程以及如何实现进程间通信(队列)

本文主要是介绍一文攻克Python的进程以及如何实现进程间通信(队列),对大家解决编程问题具有一定的参考价值,需要的程序猿们随着小编来一起学习吧!

Python的进程以及如何实现进程间通信

    • 一、进程概念
    • 二、创建进程的常用方式
      • 1、使用multiprocessing模块创建进程
      • 2、使用Process子类创建进程
      • 3、 使用进程池Pool创建进程
    • 三、通过队列实现进程间通信
      • 1、不通过其他条件,进程之间是否能共享信息?
      • 2、多线程队列的简单使用
      • 3、使用队列在进程中通信

一、进程概念

在了解进程之前,先了解多任务。多任务就是指操作系统能够执行多个任务。例如,使用Window或Linux操作系统可以同时看电影、聊天、查看网页信息等,此时操作系统就是在执行多任务,而每一个任务就是一个进程。

进程(process)是计算机已经运行程序的实体。进程与程序不同,程序本身只是指令、数据及其组织形式的描述,而进程才是(指令和数据)的真正运行实例。

二、创建进程的常用方式

1、使用multiprocessing模块创建进程

multiprocessing模块提供了一个Process类来代表一个进程对象
语法格式:

Process(group,target,name,args,kwargs)
  • group:参数未使用,值始终为None
  • target:表示当前进程启动时执行的可调用对象
  • name:为当前进程实例的别名
  • args:表示传递给target函数的参数元组
  • kwargs:表示传递给target函数的参数字典
from multiprocessing import Process

#执行子进程代码
def test():
    print("当前子进程")

#执行主程序
def main():
    print("主进程开始")
    p=Process(target=test) #实例化Process进程类
    p.start()   #启动子进程
    #p.join()    #这个是等子进程全部运行完再运行主进程
    print("主进程结束") 

if __name__=="__main__":
    main()

发现会优先执行主线程,在执行子线程
在这里插入图片描述
开启p.join()后
在这里插入图片描述

先实例化Process类,然后使用p.start()方法启动子进程开始执行test()函数Process的实例p常用的方法除start()外,还有如下常用方法:
run()如果没有给定target参数,对这个对象调用start()方法时,就将执行对象中的run()方法.
is_alive()判断进程实例是否还在执行。
join([timeout])是否等待进程实例执行结束,或等待多少秒。
start()启动进程实例(创建子进程)。
terminate()不管任务是否完成,立即终止。
Precess类还有如下常用属性:
name :当前进程实例别名,默认为Process-N,N为从1开始递增的整数。
pid :当前进程实例的PID值。

示例:创建2个子进程,分别使用os模块和time模块输出父进程和子进程的ID以及子进程的时间,并调用Process类的name和 pid属性,代码如下:

from multiprocessing import Process
import time
import os

# os.getpid()可获取当前进程id,返回值为int
# os.getppid()可获取父进程id,返回值为int

#两个子进程将会调用的俩个方法
def child_1(delay):
    print("子进程(%s)开始执行,父进程为(%s)" % (os.getpid(),os.getppid()))
    t_start = time.time()   #计时开始
    time.sleep(delay)       #程序将会被挂起delay秒
    t_end = time.time()     #即使结束
    print("子进程(%s)执行时间为%0.2f秒"%(os.getpid(),t_end-t_start))

def child_2(delay):
    print("子进程(%s)开始执行,父进程为(%s)" % (os.getpid(),os.getppid()))
    t_start = time.time()   #计时开始
    time.sleep(delay)       #程序将会被挂起delay秒
    t_end = time.time()     #即使结束
    print("子进程(%s)执行时间为%0.2f秒"%(os.getpid(),t_end-t_start))

if __name__ == "__main__":
    print("-------父进程开始执行-------")
    print("父进程PID:%s"%os.getppid)         #输出当前程序的PID
    p1=Process(target=child_1,args=(1,))    #实例化进程p1
    p2=Process(target=child_2,name="two",args=(2,)) #实例化进程p2
    p1.start()  #启动进程p1
    p2.start()  #启动进程p2

    #同时父进程仍在往下执行,如果p1、p2进程还在执行,将会返回True
    print("p1.is_alive=%s"%p1.is_alive())
    print("p2.is_alive=%s"%p2.is_alive())
    
    #输入p1和p2进程的别名和PID
    print("p1.name=%s"%p1.name)
    print("p1.pid=%s"%p1.pid)
    print("p2.name=%s"%p2.name)
    print("p2.pid=%s"%p2.pid)

    print("----------等待子进程--------")
    p1.join()        #等待p1进程结束
    p2.join()        #等待p2进程结束
    print("----------父进程执行结束---------")

在这里插入图片描述

2、使用Process子类创建进程

对于一些简单的小任务,通常使Process(target=test)方式实现多进程。但是如果要处理复杂任务的进程,通常定义一个类,使其继承Process类,每次实例化这个类的时候,就等同于实例化一个进程对象。

from multiprocessing import Process
import time 
import os

#继承Process类
class SubProcess(Process):
    #由于Process类本身也有__init__初始化方法,这个子类相当于重写了父类的这个方法
    def __init__(self,delay,name=""):
        Process.__init__(self)   #调用Process父类的初始化方法
        #self.delay相当于全局变量
        self.delay = delay       #接收参数delay
        if name:                 #判断传递的参数是否存在
            self.name = name     #如果传递参数name,则为子进程创建的name属性,否自使用默认属性
    def run(self):
        print("子进程(%s)开始执行,父进程为(%s)" % (os.getpid(),os.getppid()))
        t_start = time.time()   #计时开始
        time.sleep(self.delay)       #程序将会被挂起delay秒
        t_end = time.time()     #即使结束
        print("子进程(%s)执行时间为%0.2f秒"%(os.getpid(),t_end-t_start)) 

if __name__ == "__main__":
    print("-------父进程开始执行-------")
    print("父进程PID:%s"%os.getppid)         #输出当前程序的PID
    p1 = SubProcess(delay=1)
    p2 = SubProcess(delay=2,name = "two")

    #对一个不包含target属性的Process类执行start()方法,就会运行这个类中的run()方法
    p1.start()   #启动进程p1
    p2.start()   #启动进程p2

    #输出p1和p2进程的执行状态,如果真正执行,返回True,否则返回False
    print("p1.is_alive=%s"%p1.is_alive())
    print("p2.is_alive=%s"%p2.is_alive())
    
    #输入p1和p2进程的别名和PID
    print("p1.name=%s"%p1.name)
    print("p1.pid=%s"%p1.pid)
    print("p2.name=%s"%p2.name)
    print("p2.pid=%s"%p2.pid)

    print("----------等待子进程--------")
    p1.join()        #等待p1进程结束
    p2.join()        #等待p2进程结束
    print("----------父进程执行结束---------")

上述代码中,定义了一个SubProcess子类,继承multiprocess.Process 父类。SubProcess子类中定义了2个方法:init()初始化方法和 run()方法。在_init()初识化方法中,调用multiprocess.Process父类的_init()初始化方法,否则父类初始化方法会被覆盖,无法开启进程。此外,在 SubProcess子类中并没有定义start()方法,但在主进程中却调用了start()方法,此时就会自动执行SubPorcess类的run()方法。运行结果如图所示。
在这里插入图片描述

3、 使用进程池Pool创建进程

Pool进程池。为了更好的理解进程池,可以将进程池比作水池。我们需要完成放满10个水盆的水的任务,而在这个水池中,最多可以安放3个水盆接水,也就是同时可以执行3个任务,即开启3个进程。为更快完成任务,现在打开3个水龙头开始放水,当有一个水盆的水接满时,即该进程完成1个任务,我们就将这个水盆的水倒入水桶中,然后继续接水,即执行下一个任务。如果3个水盆每次同时装满水,那么在放满第9盆水后,系统会随机分配1个水盆接水,另外2个水盆空闲。

接下来,先来了解一下Pool类的常用方法。常用方法及说明如下:

  • apply_async(func[, args[ ,kwds]]):使用非阻塞方式调用func()函数,args为传递给func()函数的参数列表,kwds为传递给func()函数的关键字参数列表。
  • apply(func[,args[,kwds]]):使用阻塞方式调用func()函数(开行执行,堵塞方式必须等待上一个进程退出才能执行下一个进程)。
  • close():关闭Pool,使其不再接受新的任务。
  • terminate():不管任务是否完成,立即终止。
  • join():主进程阻塞,等待子进程的退出,必须在close或terminate之后使用。

关于什么是阻塞调用和非阻塞调用个人理解如下:
如果使用阻塞方式,必须等待上一个进程退出才能执行下一个进程而使用非阻塞方式,则可以并行执行3个进程。

下面通过一个示例演示一下如何使用进程池创建多进程。这里模拟水池放水的场景,定义一个进程池,设置最大进程数为3。然后使用非阻塞方式执行10个任务,查看每个进程执行的任务。具体代码:

from multiprocessing import Pool

import os,time

def task(name):
    print("子进程:%s 执行task:%s " % (os.getpid(),name))
    time.sleep(1)

if __name__ == "__main__":
    print("父进程:%s"%os.getpid())
    p = Pool(3)     #定义一个进程池,最大进程为3
    for i in range(10):
        p.apply_async(task,args=(i,)) #使用非阻塞方式调用task()函数
    print("等待所有子进程结束....")
    p.close()    #关闭进程池,关闭后p不在接收新的请求
    p.join()     #等待子进程结束
    print("所有子进程结束")

在这里插入图片描述
发现pid:3972执行了四次,而pid:15464和8704只执行了三次

阻塞方式:

from multiprocessing import Pool

import os,time

def task(name):
    print("子进程:%s 执行task:%s " % (os.getpid(),name))
    time.sleep(1)

if __name__ == "__main__":
    print("父进程:%s"%os.getpid())
    p = Pool(3)     #定义一个进程池,最大进程为3
    for i in range(10):
        # p.apply_async(task,args=(i,)) #使用非阻塞方式调用task()函数
        p.apply(task,args=(i,))         #使用阻塞方式调用task()函数
    print("等待所有子进程结束....")
    p.close()    #关闭进程池,关闭后p不在接收新的请求
    p.join()     #等待子进程结束
    print("所有子进程结束")

在这里插入图片描述

三、通过队列实现进程间通信

1、不通过其他条件,进程之间是否能共享信息?

from multiprocessing import Process
def add():
    print("......子进程1开始......")
    global num
    num += 50
    print("num:%d"%num)
    print("......子进程1结束......")

def sub():
    print("......子进程2开始......")
    global num
    num -= 50
    print("num:%d"%num)
    print("......子进程2结束......")    

num = 100 #定义一个全局变量

if __name__=="__main__":
    print(".....主进程开始.....")
    print("num:%d"%num)
    #实例化进程p1,p2
    p1 = Process(target=add)
    p2 = Process(target=sub)
    #开启进程p1,p2
    p1.start()
    p2.start()
    #阻塞主进程,等待进程结束
    p1.join()
    p2.join()
    print(".....主进程结束.....")

在这里插入图片描述
上述代码中,分别创建了2个子进程,一个子进程中令num 加上50,另一个子进程令num减去50。但是从运行结果可以看出,num在父进程和2个子进程中的初始值都是100。也就是全局变量num在一个进程中的结果,没有传递到下一个进程中,即进程之间没有共享信息。进程间示意图如图所示。
在这里插入图片描述
要如何才能实现进程间的通信呢?Python的multiprocessing模块包装了底层的机制,提供了Queue(队列)、Pipes(管道)等多种方式来交换数据。以下通过队列(Queue)来实现进程间
的通信。

2、多线程队列的简单使用

进程之间有时需要通信,操作系统提供了很多机制来实现进程间的通信。可以使用multiprocessing模块的Queue实现多进程之间的数据传递。Queue本身是一个消息列队程序,下面介绍一下Queue的使用。
初始化Queue()对象时(例如:q=Queue(num)),若括号中没有指定最大可接收的消息数量,或数量为负值,那么就代表可接受的消息数量没有上限(直到内存的尽头)。

Queue的常用方法如下;

  • Queue.qsize() :返回当前队列包含的消息数量。

  • Queue.empty():如果队列为空,返回 True;反之返回False 。

  • Queue.full():如果队列满了,返回 True;反之返回False。

  • Queue.get([block[, timeout]):获取队列中的一条消息,然后将其从列队中移除,block默认值为True。
    如果block使用默认值,且没有设置timeout(单位秒),消息列队为空,此时程序将被阻塞(停在读取状态),直到从消息列队读到消息为止。如果设置了timeout,则会等待timeout秒,若还没读取到任何消息,则抛出"Queue.Empty"异常。
    如果block值为False,消息列队为空,则会立刻抛出“Queue.Empty”异常

  • Queue.get_nowait(:相当Queue.get(False)。

  • Oueue.put(item,[block[, timeout]]):将item消息写入队列,block 默认值为True。
    如果block使用默认值,且没有设置timeout(单位秒),消息列队如果已经没有空间可写此时程序将被阻塞(停在写入状态),直到从消息列队腾出空间为止,如果设置了timeout则会等待timeout秒,若还没空间,则抛出“Queue.Full”异常。
    如果block值为False,消息列队没有空间可写入,则会立刻抛出“Queue.Full”异常

  • Oueue.put_nowait(item):相当Queue.put(item, False)。

下面,通过一个例子学习一下如何使用processing.Queue。代码如下:

from multiprocessing import Queue

if __name__=="__main__":
    q = Queue(3)         #初始化一个queue对象,最多可接收三条put消息
    q.put("消息1")
    q.put("消息2")
    print(q.full())   #q.full()  判断当前队列是否满了
    q.put("消息3")
    print(q.full())

    #因为消息队列已满,下面的try都会抛出异常
    #第一个try会等待2秒后在抛出异常,第二个try会立刻抛出异常
    try:
        q.put("消息4",True,2)
    except:
        print("消息队列已满,现有消息数量:%s"%q.qsize())
    
    try:
        q.put_nowait("消息4")
    except:
        print("消息队列已满,现有消息数量:%s"%q.qsize())

    #读取消息时,先判断消息队列是否为空,在读取
    if not q.empty():
        print("----从队列中获取消息----")
        for i in range(q.qsize()):
            print(q.get_nowait())
    
    #先判断消息队列是否已将满,在写入
    if not q.full():
        q.put_nowait("消息4")

    if not q.empty():
        print("----从队列中获取消息----")
        print(q.get_nowait())

在这里插入图片描述

3、使用队列在进程中通信

我们知道使用multiprocessing.Process 可以创建多进程,使用multiprocessing.Queue可以实现队列的操作。接下来,通过一个示例结合Process和 Queue实现进程间的通信。创建2个子进程,一个子进程负责向队列中写入数据,另一个子进程负责从队列中读取数据。为保证能够正确从队列中读取数据,设置读取数据的进程等待时间为2秒。如果2秒后仍然无法读取数据,则抛出异常。代码如下:

from multiprocessing import Process,Queue
import time

#向队列写入数据
def write_task(q):
    if not q.full():
        for i in range(5):
            message = "消息" + str(i)
            q.put(message)
            print("写入:%s"%message)

#向队列读取数据
def read_task(q):
    time.sleep(1)
    while not q.empty():
        #等待2秒,如果还没有读取到任何消息,则抛出"Queue.Empty"
        print("读取:%s"%q.get(True,2))  

if __name__=="__main__":
    print("----父进程开始----")
    #父进程创建
    q = Queue()
    pw = Process(target=write_task,args=(q,))  #实例化写入队列的子进程,并且传递队列
    pr = Process(target=read_task,args=(q,))   #实例化读取队列的子进程,并且传递队列
    pw.start()          #启动子进程pw,写入
    pr.start()          #启动子进程pr,读取
    pw.join()           #等待pw结束
    pr.join()           #等待pr结束
    print("----父进程结束----")

在这里插入图片描述

这篇关于一文攻克Python的进程以及如何实现进程间通信(队列)的文章就介绍到这儿,希望我们推荐的文章对大家有所帮助,也希望大家多多支持为之网!