Python 中通過佇列來實作行程間通信

Python程式中，在行程和行程之間是不共享全域變數的資料的，

我們來看一個例子：

from multiprocessing import Process
import os
import time

nums = [11, 22]


def work1():
    """子行程要執行的代碼"""
    print("in process1 pid=%d ,nums=%s" % (os.getpid(), nums))
    for i in range(3):
        nums.append(i)
        time.sleep(1)
        print("in process1 pid=%d ,nums=%s" % (os.getpid(), nums))


def work2():
    """子行程要執行的代碼"""
    print("in process2 pid=%d ,nums=%s" % (os.getpid(), nums))


if __name__ == '__main__':
    p1 = Process(target=work1)
    p1.start()
    p1.join()

    p2 = Process(target=work2)
    p2.start()

行程 p1 里對全域變數 nums 回圈進行處理，行程 p2 將 nums 列印出來，發現 nums 的值沒有變化，

運行結果：

in process1 pid=5788 ,nums=[11, 22]
in process1 pid=5788 ,nums=[11, 22, 0]
in process1 pid=5788 ,nums=[11, 22, 0, 1]
in process1 pid=5788 ,nums=[11, 22, 0, 1, 2]
in process2 pid=11832 ,nums=[11, 22]

通過佇列完成行程間通信

但是行程(Process)之間有時需要通信，作業系統提供了很多機制來實作行程間的通信，

可以使用 multiprocessing 模塊的 Queue 實作多行程之間的資料傳遞，

Queue 本身是一個訊息佇列程式，首先用一個小實體來演示一下 Queue 的作業原理：

from multiprocessing import Queue
# 初始化一個Queue物件，最多可接收三條put訊息
q = Queue(3)  
q.put("訊息1")
q.put("訊息2")
print(q.full())  # False
q.put("訊息3")
print(q.full())  # True

# 因為訊息佇列已滿下面的try都會拋出例外
# 第一個try會等待2秒后再拋出例外
try:
    q.put("訊息4", True, 2)
except:
    print("訊息佇列已滿，現有訊息數量:%s" % q.qsize())
# 第二個Try會立刻拋出例外
try:
    q.put_nowait("訊息4")
except:
    print("訊息列隊已滿，現有訊息數量:%s" % q.qsize())

# 推薦的方式，先判斷訊息列隊是否已滿，再寫入
if not q.full():
    q.put_nowait("訊息4")

# 讀取訊息時，先判斷訊息列隊是否為空，再讀取
if not q.empty():
    for i in range(q.qsize()):
        print(q.get_nowait())

運行結果：

佇列 Queue 的使用說明

初始化 Queue()物件時（例如：q=Queue()），若括號中沒有指定最大可接收的訊息數量，或數量為負值，那么就代表可接受的訊息數量沒有上限（直到記憶體的盡頭），

Queue.qsize()：回傳當前佇列包含的訊息數量，

Queue.empty()：如果佇列為空，回傳True，反之False，

Queue.full()：如果佇列滿了，回傳True，反之False，

Queue.get([block[, timeout]])：獲取佇列中的一條訊息，然后將其從列隊中移除，block 默認值為 True，

1. 如果 block 使用默認值，且沒有設定 timeout（單位秒），訊息佇列如果為空，此時程式將被阻塞，停在讀取狀態，直到從訊息佇列讀到訊息為止；如果設定了 timeout，則會等待 timeout 秒，若還沒讀取到任何訊息，則拋出 "Queue.Empty" 例外，
2. 如果 block 值為 False，訊息列隊如果為空，則會立刻拋出 "Queue.Empty" 例外，

Queue.get_nowait()：相當 Queue.get(False)，

Queue.put(item,[block[, timeout]])：將 item 訊息寫入佇列，block 默認值為 True，

1. 如果 block 使用默認值，且沒有設定 timeout（單位秒），訊息佇列如果已經沒有空間可寫入，此時程式將被阻塞，停在寫入狀態，直到從訊息佇列騰出空間為止；如果設定了timeout，則會等待 timeout 秒，若還沒空間，則拋出 "Queue.Full" 例外，
2. 如果 block 值為 False，訊息佇列如果沒有空間可寫入，則會立刻拋出 "Queue.Full" 例外，

Queue.put_nowait(item)：相當Queue.put(item, False)，

Queue實體

我們以 Queue 為例，在父行程中創建兩個子行程，一個往 Queue 里寫資料，一個從 Queue 里讀資料，

from multiprocessing import Process, Queue
import os
import time
import random


def write(q):
    # 寫資料行程執行的代碼:
    for value in ['A', 'B', 'C']:
        print('Put %s to queue...' % value)
        q.put(value)
        time.sleep(random.random())


def read(q):
    # 讀資料行程執行的代碼:
    while True:
        if not q.empty():
            value = https://www.cnblogs.com/studyming/p/q.get(True)
            print('Get %s from queue.' % value)
            time.sleep(random.random())
        else:
            break


if __name__ == '__main__':
    # 父行程創建Queue，并傳給各個子行程：
    q = Queue()
    pw = Process(target=write, args=(q,))
    pr = Process(target=read, args=(q,))
    # 啟動子行程pw，寫入:
    pw.start()
    # 等待pw結束:
    pw.join()
    # 啟動子行程pr，讀取:
    pr.start()
    pr.join()
    print('')
    print('所有資料都寫入并且讀完')

運行結果：

標籤：Python

上一篇：17-用python爬取下載女神照片

下一篇：python selenium click打開新視窗