Python多進程開發(fā)之如何輕松突破GIL瓶頸

更新時間：2025年06月23日 08:11:55 作者：燭陰

這篇文章主要來和大家詳細介紹一下Python多進程開發(fā)的相關知識,看看它是如何繞過GIL全局解釋器鎖的,感興趣的小伙伴可以跟隨小編一起學習一下

1. 為啥要用多進程？先認清GIL

Python多線程為何沒法加速CPU密集型任務？因為GIL（全局解釋器鎖）：在主流的CPython解釋器中，無論多少線程歸根結底同時只有一個線程在執(zhí)行Python字節(jié)碼，嚴重影響CPU計算性能的并行。

多進程機制完美繞開了GIL：每個進程擁有自己的Python解釋器和內存空間，真正多核、多CPU并行執(zhí)行，CPU密集型任務大大提速！

2. Python多進程基礎用法：multiprocessing庫

Python 內置標準庫 multiprocessing，API與threading非常類似，上手容易。

2.1 Proces介紹

Process([group [, target [, name [, args [, kwargs]]]]])

target 表示調用對象
args 表示調用對象的位置參數(shù)元組
kwargs 表示調用對象的字典
name 為別名
group 實質上不使用

2.2 Hello，多進程

from multiprocessing import Process
import os

def worker():
    print('子進程PID:', os.getpid())

if __name__ == '__main__':
    p = Process(target=worker)
    p.start()
    p.join()
    print('主進程結束')

注意！ Windows下多進程要寫在 if __name__ == '__main__': 保護塊內，否則會無限遞歸啟動進程。

2.3 多進程并行計算

from multiprocessing import Process
import time

def task(name):
    print(f'{name} 開始')
    time.sleep(2)
    print(f'{name} 結束')

if __name__ == '__main__':
    proc_list = []
    for i in range(3):
        p = Process(target=task, args=(f'進程{i+1}',))
        p.start()
        proc_list.append(p)
    for p in proc_list:
        p.join()
    print('所有子進程結束')

多個進程可并發(fā)分擔任務，真正多核利用！

2.4 子進程類實現(xiàn)（面向對象）

import multiprocessing
import time

class ClockProcess(multiprocessing.Process):
    def __init__(self, interval):
        multiprocessing.Process.__init__(self)
        self.interval = interval

    def run(self):
        n = 5
        while n > 0:
            print("當前時間: {0}".format(time.ctime()))
            time.sleep(self.interval)
            n -= 1


if __name__ == '__main__':
    p = ClockProcess(3)
    p.start()

3. 進階：進程池批量并行(Process Pool)

大量任務并發(fā)時，手動管理進程很麻煩，用Pool池自動分配和收集結果，極大提升效率！

from multiprocessing import Pool
import os, time

def square(n):
    print('計算', n, '在進程', os.getpid())
    time.sleep(1)
    return n * n

if __name__ == '__main__':
    with Pool(processes=3) as pool:
        results = pool.map(square, range(5))
    print('返回結果:', results)

比循環(huán)創(chuàng)建 Process 簡潔得多，還能自動分配CPU，每個子進程并發(fā)執(zhí)行給定任務。

4. 多進程間數(shù)據(jù)：通信與共享

每個進程獨立內存，不能像多線程那樣直接共享變量
多進程用隊列（Queue）、管道（Pipe）或特殊共享對象實現(xiàn)

4.1 用隊列 Queue 通信（推薦）

from multiprocessing import Process, Queue

def producer(q):
    for i in range(5):
        q.put(i)
    q.put(None)  # 發(fā)送結束標記

def consumer(q):
    while True:
        item = q.get()
        if item is None:
            break
        print('消費:', item)

if __name__ == '__main__':
    q = Queue()
    p1 = Process(target=producer, args=(q,))
    p2 = Process(target=consumer, args=(q,))
    p1.start()
    p2.start()
    p1.join()
    p2.join()

4.2 共享內存變量：Value 與 Array

from multiprocessing import Process, Value, Array

def worker(val, arr):
    val.value += 10
    for i in range(len(arr)):
        arr[i] *= 2

if __name__ == '__main__':
    num = Value('i', 5)                # int類型共享變量
    arr = Array('i', [1, 2, 3, 4])     # 數(shù)組
    p = Process(target=worker, args=(num, arr))
    p.start()
    p.join()
    print('num =', num.value)
    print('arr =', list(arr))

5. 多進程安全：鎖機制

多進程也需防止競爭條件，通過 Lock/Manager 加鎖同步。

from multiprocessing import Process, Lock, Value

def add(lock, counter):
    for _ in range(100000):
        with lock:
            counter.value += 1

if __name__ == '__main__':
    lock = Lock()
    counter = Value('i', 0)
    procs = [Process(target=add, args=(lock, counter)) for _ in range(4)]
    for p in procs:
        p.start()
    for p in procs:
        p.join()
    print('結果:', counter.value)