快捷導(dǎo)航

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

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

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

1. 為啥要用多進(jìn)程？先認(rèn)清GIL

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

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

2. Python多進(jìn)程基礎(chǔ)用法：multiprocessing庫

Python 內(nèi)置標(biāo)準(zhǔn)庫 multiprocessing，API與threading非常類似，上手容易。

2.1 Proces介紹

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

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

2.2 Hello，多進(jìn)程

from multiprocessing import Process
import os

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

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

注意！ Windows下多進(jìn)程要寫在 if __name__ == '__main__': 保護(hù)塊內(nèi)，否則會(huì)無限遞歸啟動(dòng)進(jìn)程。

2.3 多進(jìn)程并行計(jì)算

from multiprocessing import Process
import time

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

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

多個(gè)進(jìn)程可并發(fā)分擔(dān)任務(wù)，真正多核利用！

2.4 子進(jìn)程類實(shí)現(xiàn)（面向?qū)ο螅?/h3>

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("當(dāng)前時(shí)間: {0}".format(time.ctime()))
            time.sleep(self.interval)
            n -= 1


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

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

大量任務(wù)并發(fā)時(shí)，手動(dòng)管理進(jìn)程很麻煩，用Pool池自動(dòng)分配和收集結(jié)果，極大提升效率！

from multiprocessing import Pool
import os, time

def square(n):
    print('計(jì)算', n, '在進(jì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('返回結(jié)果:', results)

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

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

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

4.1 用隊(duì)列 Queue 通信（推薦）

from multiprocessing import Process, Queue

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

def consumer(q):
    while True:
        item = q.get()
        if item is None:
            break
        print('消費(fèi):', 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 共享內(nèi)存變量：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. 多進(jìn)程安全：鎖機(jī)制

多進(jìn)程也需防止競(jìng)爭(zhēng)條件，通過 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('結(jié)果:', counter.value)