1. 進程

1.1 進程的創建 fork

正在運行著的代碼,就稱為進程

# 示例:
import os

# 注意: fork 函數,只在 Unix/Linux/Mac 上運行, windows 不可以
pid = os.fork()
print(pid)
if pid == 0:
    # 子進程執行(pid==0)
    print("=== 子進程 === %d %d" % (os.getpid(),os.getppid()))
else:
    # 主(父)進程執行(pid>0)
    print("=== 父進程 ===%d" % os.getpid())

1.2 全局變量在進程中不共享

# 示例:
import os
import time

g_num = 100

ret = os.fork()
if ret == 0:
    print("=== process 1 ===")
    g_num += 1
    print("=== process 1  g_num = %d===" % g_num)
else:
    time.sleep(3)
    print("=== process 2 ===")
    print("=== process 2 g_num = %d ===" % g_num)

# 輸出:
=== process 1 ===
=== process 1  g_num = 101 ===
=== process 2 ===
=== process 2  g_num = 100 ===

1.3 多次fork

# 示例:
import os

ret = os.fork()
if ret == 0:
    print("=== 1 ===")
else:
    print("=== 2 ===")

ret = os.fork()
if ret == 0:
    print("=== 3 ===")
else:
    print("=== 4 ===")

# 輸出:
=== 2 ===
=== 4 ===
=== 1 ===
=== 3 ===
=== 4 ===
=== 3 ===

1.4 Process 創建子進程

# 示例
from multiprocessing import Process
import os

# 子進程要執行的代碼
def run_proc(name):
    print('子進程運行中, name=%s, pid=%d...' % (name, os.getpid()))


if __name__=='__main__':
    print('父進程 %d.' % os.getpid())
    p = Process(target=run_proc, args=('test',))
    print('子進程將要執行')
    p.start()   # 創建子進程
    p.join()    # 等到子進程實例(p)執行結束后,才執行后面的代碼
    print('子進程已結束')

1.5 Process 的子類

# 示例:
from multiprocessing import Process
import time
import os

# 繼承Process類
class Process_Class(Process):
    # 因為Process類本身也有 __init__ 方法,這個類相當于重寫了這個方法;
    # 但這樣就會帶來一個問題,我們并沒有完全的初始化一個 Process 類, 所以就不能使用從這個類繼承的一些方法
    # 最好的方法就是將繼承類本身傳遞給 Process.__init__ 方法, 完成這些初始化操作
    def __init__(self, interval):
        # super(Process_Class, self).__init__()
        Process.__init__(self)
        self.interval = interval

    # 重寫Process類的run()方法
    def run(self):
        print("子進程(%s) 開始執行, 父進程為 (%s)" % (os.getpid(), os.getppid()))
        t_start = time.time()
        time.sleep(self.interval)
        t_stop = time.time()
        print("(%s)執行結束, 耗時%0.2f秒" % (os.getpid(), t_stop-t_start))

if __name__=="__main__":
    t_start = time.time()
    print("當前進程(%s)" % os.getpid())
    p1 = Process_Class(2)
    # 對一個不包含target屬性的Process類執行start()方法,就會運行這個類中的 run() 方法
    p1.start()
    p1.join()
    t_stop = time.time()
    print("(%s)執行結束, 耗時%0.2f" % (os.getpid(), t_stop-t_start))

1.6 進程池 Pool(非阻塞方式)

# 示例:
from multiprocessing import Pool
import os, time, random

def worker(msg):
    t_start = time.time()
    print("%s開始執行, 進程號為%d" % (msg, os.getpid()))
    # random.random()隨機生成0~1之間的浮點數
    time.sleep(random.random()*2)
    t_stop = time.time()
    print(msg, "執行完畢, 耗時%0.2f" % (t_stop-t_start))

po = Pool(3)  # 定義一個進程池, 最大進程數 3
for i in range(0, 10):
    # Pool.apply_async(要調用的目標,(傳遞給目標的參數元組,))
    # 每次循環將會用空閑出來的子進程去調用目標
    po.apply_async(worker,(i,))

print("=== start ===")
po.close()  # 關閉進程池,關閉后po不再接收新的請求
po.join()   # 等待po中所有子進程執行完成, 必須放在close語句之后
print("=== close ===")

1.7 apply 阻塞方式添加任務

# 示例:
from multiprocessing import Pool
import os, time, random

def worker(msg):
    t_start = time.time()
    print("%s開始執行, 進程號為%d" % (msg, os.getpid()))
    # random.random()隨機生成0~1之間的浮點數
    time.sleep(random.random()*2)
    t_stop = time.time()
    print(msg, "執行完畢, 耗時%0.2f" % (t_stop-t_start))

po = Pool(3)  # 定義一個進程池, 最大進程數 3
for i in range(0, 10):
    po.apply(worker, (i,))

print("=== start ===")
po.close()
po.join()
print("=== close ===")

1.8 進程間通信(Queue)

# 示例一: Queue 存取數據
from multiprocessing import Queue
q=Queue(3)  # 初始化一個Queue對象,最多可接收三條put消息
q.put("消息1")
q.put("消息2")
print(q.full())     # False
q.put("消息3")
print(q.full())     # True

# 因為消息隊列已滿,下面的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.full():
    q.put_nowait("消息4")

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



# 示例二: 進程間通信
# 在父進程中創建兩個子進程,一個往Queue里寫數據, 一個從Queue里讀數據
from multiprocessing import Process, Queue
import os, time, 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 = 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('所有數據都寫入并讀完')

1.8.1 進程間通信(Pipe)

# 示例:
from multiprocessing import Process, Pipe

def f(conn):
    conn.send([33, None, 'hello'])
    conn.close()

if __name__ = '__main__':
    parent_conn, child_conn = Pipe()
    p = Process(target = f, args = (child_conn, ))
    p.start()
    print(parent_conn.recv())
    p.join()

1.9 進程池中的Queue

在進程池中,需要使用 multiprocessing.Manager()中的 Queue();而不是 multiprocessing.Queue()

# 示例:
from multiprocessing import Manager, Pool
import os, time, random


# 寫數據進程執行代碼
def reader(q):
    print("reader啟動(%s), 父進程為(%s)" % (os.getpid(), os.getppid()))
    for i in range(q.qsize()):
        print("reader從Queue獲取到消息: %s" %q.get(True))

# 讀數據進程執行代碼:
def writer(q):
    print("writer啟動(%s), 父進程為(%s)" % (os.getpid(), os.getppid()))
    for i in ['A', 'B', 'C']:
        q.put(i)

if __name__=="__main__":
    print("(%s) start" % os.getpid())
    q=Manager().Queue()     # 使用Manager中的Queue來初始化
    po=Pool()
    # 使用阻塞模式創建進程,這樣就不需要在reader中使用死循環了, 可以讓writer執行完成后,再用reader讀取
    po.apply(writer,(q,))
    po.apply(reader,(q,))
    po.close()
    po.join()
    print("(%s) End" % os.getpid())

參考資料:

Python 核心編程

總結

以上是生活随笔為你收集整理的Python 进程(process)的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。