Python 的標準直譯器（CPython）中有一個稱為 GIL 的機制，即使使用多執行緒，也無法同時執行多個執行緒。因此，若想充分利用 CPU，建議使用多進程。
1.3 使用 Python 實作簡單的多進程範例
import multiprocessing
import time
def worker(n):
print(f"プロセス {n} 開始")
time.sleep(2)
print(f"プロセス {n} 終了")
if __name__ == "__main__":
process_list = []
# 3つのプロセスを作成
for i in range(3):
p = multiprocessing.Process(target=worker, args=(i,))
process_list.append(p)
p.start()
# 全てのプロセスが終了するまで待機
for p in process_list:
p.join()
print("全プロセス終了")
1.4 使用多進程時的注意事項
1. 在 Windows 系統上必須加上 if __name__ == "__main__":
在 Windows 環境中，使用 multiprocessing.Process() 時，若未加上 if __name__ == "__main__": 會發生錯誤。
錯誤的範例（會出現錯誤）
import multiprocessing
def worker():
print("Hello from process")
p = multiprocessing.Process(target=worker)
p.start()
此程式碼在 Windows 系統中會報錯。
正確的範例
import multiprocessing
def worker():
print("Hello from process")
if __name__ == "__main__":
p = multiprocessing.Process(target=worker)
p.start()
p.join()
加上 if __name__ == "__main__": 即可在 Windows 上正確執行。
1.5 小結
什麼是多進程？ → 同時執行多個進程的方式
與多執行緒的差異 → 不受 GIL 限制，適合 CPU 密集型任務
在 Python 中的簡單範例 → 使用 multiprocessing.Process()
Windows 注意事項 → 必須加上 if __name__ == "__main__":
2. 實戰篇：multiprocessing 模組的使用方式
2.1 multiprocessing 模組的概要
multiprocessing 模組是 Python 的標準函式庫，提供 基於進程的平行處理 功能。
透過這個模組，可以充分運用 CPU 核心，並避開 GIL 的限制。
import multiprocessing
def square(n):
return n * n
if __name__ == "__main__":
with multiprocessing.Pool(4) as pool:
results = pool.map(square, range(10))
print(results)
2.6 小結
使用 multiprocessing 模組可以 輕鬆實作平行處理
使用 Process 類別建立個別進程
使用 Queue 或 Pipe 可以 在進程之間傳遞資料
透過 Value 或 Array 可以 實現共享記憶體
使用 Pool 類別可以 高效處理大量資料
3. 進階篇：錯誤處理與效能最佳化
3.1 在 multiprocessing 中常見的錯誤與對策
錯誤 1：Windows 環境中未加上 if __name__ == "__main__": 的錯誤
錯誤訊息
RuntimeError: freeze_support() must be called if program is run in frozen mode
解決方法
import multiprocessing
def worker():
print("Hello from process")
if __name__ == "__main__": # 這一行是必要的
p = multiprocessing.Process(target=worker)
p.start()
p.join()
錯誤 2：PicklingError（無法在進程間傳遞函式）
錯誤訊息
AttributeError: Can't pickle local object 'main.<locals>.<lambda>'
解決方法
import multiprocessing
def square(x): # 使用全域函式
return x * x
if __name__ == "__main__":
with multiprocessing.Pool(4) as pool:
results = pool.map(square, range(10)) # 避免使用 lambda
print(results)
import multiprocessing
def worker(n):
return n * n
if __name__ == "__main__":
num_workers = multiprocessing.cpu_count() # 取得 CPU 核心數量
with multiprocessing.Pool(num_workers) as pool:
results = pool.map(worker, range(100))
print(results)
最佳化 2：使用 Pool.starmap()
import multiprocessing
def multiply(a, b):
return a * b
if __name__ == "__main__":
with multiprocessing.Pool(4) as pool:
results = pool.starmap(multiply, [(1, 2), (3, 4), (5, 6)])
print(results)
import multiprocessing
def worker(n):
return n * n
if __name__ == "__main__":
with multiprocessing.Pool(multiprocessing.cpu_count()) as pool:
results = pool.map(worker, range(100))
print(results)
4.3 如何在多進程中共享字典或列表？
解答
可以使用 multiprocessing.Manager() 來實現共享物件
import multiprocessing
def worker(shared_list):
shared_list.append(100) # 更新共享列表
if __name__ == "__main__":
with multiprocessing.Manager() as manager:
shared_list = manager.list([1, 2, 3])
p = multiprocessing.Process(target=worker, args=(shared_list,))
p.start()
p.join()
print(shared_list) # [1, 2, 3, 100]
