Python中進(jìn)程間通信的示例分析

這篇文章給大家分享的是有關(guān)Python中進(jìn)程間通信的示例分析的內(nèi)容。小編覺得挺實(shí)用的，因此分享給大家做個(gè)參考，一起跟隨小編過來看看吧。

進(jìn)程概述

進(jìn)程（Process）是計(jì)算機(jī)中已運(yùn)行程序的實(shí)體。進(jìn)程與程序不同，程序本身只是指令、數(shù)據(jù)及器組織形式的描述，進(jìn)程才是程序（那些指令和數(shù)據(jù)）的真正運(yùn)行實(shí)體。例如在沒有打開QQ時(shí)，QQ只是程序。打開以后，操作系統(tǒng)為QQ開啟一個(gè)進(jìn)程。再打開一個(gè)QQ，則又開啟一個(gè)進(jìn)程。

那么在多進(jìn)程中，每個(gè)進(jìn)程之間是什么關(guān)系呢？其實(shí)每個(gè)進(jìn)程都有自己的地址空間、內(nèi)存、數(shù)據(jù)棧以及其他記錄其運(yùn)行狀態(tài)的輔助數(shù)據(jù)。下通過一個(gè)例子驗(yàn)證一下進(jìn)程間是否能直接共享信息。示例代碼如下：

from multiprocessing import Process

def plus():
    print('-------子進(jìn)程1開始------')
    global g_num
    g_num += 50
    print('g_num is %d'%g_num)
    print('-------子進(jìn)程1結(jié)束------')

def minus():
    print('-------子進(jìn)程2開始------')
    global g_num
    g_num -= 50
    print('g_num is %d'%g_num)
    print('-------子進(jìn)程2結(jié)束------')

g_num = 100 # 定義一個(gè)全局變量
if __name__ == '__main__':
    print('-------主進(jìn)程開始------')
    print('g_num is %d'%g_num)
    p1 = Process(target=plus)   # 實(shí)例化進(jìn)程p1
    p2 = Process(target=minus)  # 實(shí)例化進(jìn)程p2
    p1.start()                  # 開啟進(jìn)程p1
    p2.start()                  # 開啟進(jìn)程p2
    p1.join()                   # 等待p1進(jìn)程結(jié)束
    p2.join()                   # 等待p2進(jìn)程結(jié)束
    print('-------主進(jìn)程結(jié)束------')

示例代碼中定義一個(gè)全局變量g_num,分別創(chuàng)建2個(gè)子進(jìn)程對(duì)g_num變量執(zhí)行不同的操作，并輸出操作后的結(jié)果。運(yùn)行結(jié)果如下：

-------主進(jìn)程開始------
g_num is 100
-------子進(jìn)程1開始------
g_num is 150
-------子進(jìn)程1結(jié)束------
-------子進(jìn)程2開始------
g_num is 50
-------子進(jìn)程2結(jié)束------
-------主進(jìn)程結(jié)束------

Process finished with exit code 0

上述代碼中，分別創(chuàng)建了2個(gè)子進(jìn)程，一個(gè)子進(jìn)程中令g_num變量加50，另一個(gè)子進(jìn)程令g_num變量減50。但是從運(yùn)行結(jié)果看，g_num變量在父進(jìn)程和2個(gè)子進(jìn)程中的初識(shí)值都是100，也就是說全局變量g_num在一個(gè)進(jìn)程中的結(jié)果并沒有傳到下一個(gè)進(jìn)程中，即進(jìn)程之間并沒有共享信息。

要如何才能實(shí)現(xiàn)進(jìn)程間的通信呢？Python的multiprocessing模塊包裝了底層的機(jī)制，提供了Queue(隊(duì)列)、Pipes(管道)等多種方式來交換數(shù)據(jù)。

隊(duì)列簡(jiǎn)介

隊(duì)列（Queue)就是模仿現(xiàn)實(shí)中的排隊(duì)。舉個(gè)栗子（非網(wǎng)上購(gòu)票方式，曾經(jīng)的買電影票的方式），例如排隊(duì)買電影票，新來的人排到隊(duì)伍最后，最前面的人買完票走開，后面的人跟上。由此可見隊(duì)列的兩個(gè)特點(diǎn)：

§ 新來的都排在隊(duì)尾

§ 最前面的完成后離隊(duì)，后面一個(gè)跟上

多進(jìn)程隊(duì)列的使用

進(jìn)程間有時(shí)需要通信，操作系統(tǒng)提供了很多機(jī)制來實(shí)現(xiàn)進(jìn)程間的通信，如可以使用multiprocessing模塊的Queue隊(duì)列實(shí)現(xiàn)多進(jìn)程之間的數(shù)據(jù)傳遞。Queue本身是一個(gè)消息隊(duì)列程序，下面介紹一下它的使用。

初始化Queue()對(duì)象時(shí)（例如：q=Queue(num))，若括號(hào)中沒有指定最大可接收的消息數(shù)量，或數(shù)量為負(fù)值，那么就代表可接受的消息數(shù)量沒有上限（直到內(nèi)存的盡頭）。Queue常用方法如下：

§ Queue.qsize():返回當(dāng)前隊(duì)列包含的消息數(shù)量

§ Queue.empty():如果隊(duì)列為空，返回True,否則返回False

§ Queue.full():如果隊(duì)列滿了，返回True,否則返回False

§ Queue.get([block[,timeout]]):獲取隊(duì)列中的一條消息，然后將其從隊(duì)列中移除，block默認(rèn)值為True

如果block使用默認(rèn)值，且沒有設(shè)置timeout(單位秒)，消息隊(duì)列為空，此時(shí)程序?qū)⒈蛔枞ㄍＴ谧x取狀態(tài)），直到從消息隊(duì)列中讀到消息為止，如果設(shè)置了timeou，則會(huì)等待timeout秒，若還沒有讀取到任何消息，則拋出“Queue.Empty“異常
如果block值為False，消息隊(duì)列為空，則會(huì)立刻拋出“Queue.Empty“異常

§ Queue.get_nowait():相當(dāng)Queue.get(Flase)

§ Queue.put(item,[block[,timeout]]):將item消息寫入隊(duì)列，block默認(rèn)值為True

如果block使用默認(rèn)值，且沒有設(shè)置timeout（單位秒），當(dāng)消息隊(duì)列已經(jīng)沒有空間可寫入時(shí)，程序?qū)⒈蛔枞ㄍＴ趯懭霠顟B(tài)），直到從消息隊(duì)列騰出空間為止，如果設(shè)置了timeout，則會(huì)等待timeout秒，若還沒有空間，則拋出“Queue.Full“異常
如果block值為False,當(dāng)消息隊(duì)列沒有空間可寫入時(shí)，則會(huì)立刻拋出“Queue.Full“異常
Queue.put_nowait(item):相當(dāng)Queue.put(item,False)

示例代碼如下：

#coding=utf-8
from multiprocessing import Queue

if __name__ == '__main__':
    q=Queue(3) # 初始化一個(gè)Queue對(duì)象，最多可接收三條put消息
    q.put("消息1")
    q.put("消息2")
    print(q.full())  # 返回False
    q.put("消息3")
    print(q.full()) # 返回True

    # 因?yàn)橄㈥?duì)列已滿，下面的try會(huì)拋出異常，
    # 第一個(gè)try會(huì)等待2秒后再拋出異常，第二個(gè)try會(huì)立刻拋出異常
    try:
        q.put("消息4",True,2)
    except:
        print("消息隊(duì)列已滿，現(xiàn)有消息數(shù)量:%s"%q.qsize())

    try:
        q.put_nowait("消息4")
    except:
        print("消息隊(duì)列已滿，現(xiàn)有消息數(shù)量:%s"%q.qsize())

    # 讀取消息時(shí)，先判斷消息隊(duì)列是否為空，為空時(shí)再讀取
    if not q.empty():
        print('----從隊(duì)列中獲取消息---')
        for i in range(q.qsize()):
            print(q.get_nowait())
    # 先判斷消息隊(duì)列是否已滿，不為滿時(shí)再寫入
    if not q.full():
        q.put_nowait("消息4")

程序運(yùn)行結(jié)果如下：

False
True
消息隊(duì)列已滿，現(xiàn)有消息數(shù)量:3
消息隊(duì)列已滿，現(xiàn)有消息數(shù)量:3
----從隊(duì)列中獲取消息---
消息1
消息2
消息3

備 注

此程序只能在Windows環(huán)境下運(yùn)行成功，mac系統(tǒng)會(huì)報(bào)錯(cuò)。不知道什么原因？而且單獨(dú)的print(q.qsize())都報(bào)錯(cuò)。

使用隊(duì)列在進(jìn)程間通信

我們知道使用multiprocessing.Process可以創(chuàng)建多進(jìn)程，使用multiprocessing.Queue可以實(shí)現(xiàn)隊(duì)列的操作。結(jié)合Process和Queue實(shí)現(xiàn)進(jìn)程間的通信。示例代碼如下：

from multiprocessing import Process, Queue
import  time

# 向隊(duì)列中寫入數(shù)據(jù)
def write_task(q):
    if not q.full():
        for i in range(5):
            message = "消息" + str(i)
            q.put(message)
            print("寫入:%s"%message)
# 從隊(duì)列讀取數(shù)據(jù)
def read_task(q):
    time.sleep(1)                      # 休眠1秒
    while not q.empty():
        print("讀取:%s" % q.get(True,2))     # 等待2秒，如果還沒讀取到任何消息，
                                           # 則拋出"Queue.Empty"異常

if __name__ == "__main__":
    print("-----父進(jìn)程開始-----")
    q = Queue()  # 父進(jìn)程創(chuàng)建Queue，并傳給各個(gè)子進(jìn)程
    pw = Process(target=write_task, args=(q,)) # 實(shí)例化寫入隊(duì)列的子進(jìn)程，并且傳遞隊(duì)列
    pr = Process(target=read_task, args=(q,))  # 實(shí)例化讀取隊(duì)列的子進(jìn)程，并且傳遞隊(duì)列
    pw.start()   # 啟動(dòng)子進(jìn)程 pw，寫入
    pr.start()   # 啟動(dòng)子進(jìn)程 pr，讀取
    pw.join()    # 等待 pw 結(jié)束
    pr.join()    # 等待 pr 結(jié)束
    print("-----父進(jìn)程結(jié)束-----")

上述代碼中創(chuàng)建2個(gè)子進(jìn)程，一個(gè)子進(jìn)程負(fù)責(zé)向隊(duì)列中寫入數(shù)據(jù)，另一個(gè)子進(jìn)程負(fù)責(zé)從隊(duì)列中讀取數(shù)據(jù)。為保證能夠正確從隊(duì)列中讀取數(shù)據(jù)，設(shè)置讀取數(shù)據(jù)的進(jìn)程等待時(shí)間為2秒。如果2秒后仍然無(wú)法讀取數(shù)據(jù)，則拋出異常。運(yùn)行結(jié)果如下：

-----父進(jìn)程開始-----
寫入:消息0
寫入:消息1
寫入:消息2
寫入:消息3
寫入:消息4
讀取:消息0
讀取:消息1
讀取:消息2
讀取:消息3
讀取:消息4
-----父進(jìn)程結(jié)束-----

Process finished with exit code 0

感謝各位的閱讀！關(guān)于“Python中進(jìn)程間通信的示例分析”這篇文章就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，讓大家可以學(xué)到更多知識(shí)，如果覺得文章不錯(cuò)，可以把它分享出去讓更多的人看到吧！