Python程序中，在進(jìn)程和進(jìn)程之間是不共享全局變量的數(shù)據(jù)的。

我們來(lái)看一個(gè)例子：

from multiprocessing import Process
import os
import time

nums = [11, 22]


def work1():
  """子進(jìn)程要執(zhí)行的代碼"""
  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():
  """子進(jìn)程要執(zhí)行的代碼"""
  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()

進(jìn)程 p1 里對(duì)全局變量 nums 循環(huán)進(jìn)行處理，進(jìn)程 p2 將 nums 打印出來(lái)，發(fā)現(xiàn) nums 的值沒(méi)有變化。

運(yùn)行結(jié)果：

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]

通過(guò)隊(duì)列完成進(jìn)程間通信

但是進(jìn)程(Process)之間有時(shí)需要通信，操作系統(tǒng)提供了很多機(jī)制來(lái)實(shí)現(xiàn)進(jìn)程間的通信。

可以使用 multiprocessing 模塊的 Queue 實(shí)現(xiàn)多進(jìn)程之間的數(shù)據(jù)傳遞。

Queue 本身是一個(gè)消息隊(duì)列程序，首先用一個(gè)小實(shí)例來(lái)演示一下 Queue 的工作原理：

from multiprocessing import Queue
# 初始化一個(gè)Queue對(duì)象，最多可接收三條put消息
q = Queue(3) 
q.put("消息1")
q.put("消息2")
print(q.full()) # False
q.put("消息3")
print(q.full()) # True

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

# 推薦的方式，先判斷消息列隊(duì)是否已滿(mǎn)，再寫(xiě)入
if not q.full():
  q.put_nowait("消息4")

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

運(yùn)行結(jié)果：

隊(duì)列 Queue 的使用說(shuō)明

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

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

Queue.empty()：如果隊(duì)列為空，返回True，反之False。

Queue.full()：如果隊(duì)列滿(mǎn)了，返回True，反之False。

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

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

Queue.get_nowait()：相當(dāng) Queue.get(False)。

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

1. 如果 block 使用默認(rèn)值，且沒(méi)有設(shè)置 timeout（單位秒），消息隊(duì)列如果已經(jīng)沒(méi)有空間可寫(xiě)入，此時(shí)程序?qū)⒈蛔枞Ｔ趯?xiě)入狀態(tài)，直到從消息隊(duì)列騰出空間為止；如果設(shè)置了timeout，則會(huì)等待 timeout 秒，若還沒(méi)空間，則拋出 "Queue.Full" 異常。
2. 如果 block 值為 False，消息隊(duì)列如果沒(méi)有空間可寫(xiě)入，則會(huì)立刻拋出 "Queue.Full" 異常。
   

Queue.put_nowait(item)：相當(dāng)Queue.put(item, False)。

Queue實(shí)例

我們以 Queue 為例，在父進(jìn)程中創(chuàng)建兩個(gè)子進(jìn)程，一個(gè)往 Queue 里寫(xiě)數(shù)據(jù)，一個(gè)從 Queue 里讀數(shù)據(jù)。

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


def write(q):
  # 寫(xiě)數(shù)據(jù)進(jìn)程執(zhí)行的代碼:
  for value in ['A', 'B', 'C']:
    print('Put %s to queue...' % value)
    q.put(value)
    time.sleep(random.random())


def read(q):
  # 讀數(shù)據(jù)進(jìn)程執(zhí)行的代碼:
  while True:
    if not q.empty():
      value = q.get(True)
      print('Get %s from queue.' % value)
      time.sleep(random.random())
    else:
      break


if __name__ == '__main__':
  # 父進(jìn)程創(chuàng)建Queue，并傳給各個(gè)子進(jìn)程：
  q = Queue()
  pw = Process(target=write, args=(q,))
  pr = Process(target=read, args=(q,))
  # 啟動(dòng)子進(jìn)程pw，寫(xiě)入:
  pw.start()
  # 等待pw結(jié)束:
  pw.join()
  # 啟動(dòng)子進(jìn)程pr，讀取:
  pr.start()
  pr.join()
  print('')
  print('所有數(shù)據(jù)都寫(xiě)入并且讀完')

運(yùn)行結(jié)果：

以上就是Python通過(guò)隊(duì)列來(lái)實(shí)現(xiàn)進(jìn)程間通信的示例的詳細(xì)內(nèi)容，更多關(guān)于python實(shí)現(xiàn)進(jìn)程間通信的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

最新評(píng)論