Process, Thread, Coroutine，A Hands-on Experiment to Conquer Concurrency in Python and Golang

import time
import os
from multiprocessing import Process
from threading import Thread
import asyncio

# --- 參數設定 ---
# 測試執行緒/行程/協程的數量
NUM_WORKERS = 8
# CPU 密集型任務的強度 (數值越大，運算時間越久，請根據您的電腦性能調整)
N_CPU = 2
# I/O 密集型任務的模擬等待時間
DELAY_IO = 0.5

# --- 任務定義 ---

# 1. CPU 密集型任務：耗時的計算
def cpu_bound_task(n):
    """佔用 CPU 的運算任務"""
    i = 0
    while i < n * 500000:
        i += 1
        # 進行簡單的數學運算以佔用 CPU
        x = 51000**0.5 
    print(f"[{os.getpid()}] CPU-Bound Task finished.")

# 2. I/O 密集型任務：模擬網路等待
def io_bound_task(delay):
    """模擬網路請求，等待指定時間"""
    time.sleep(delay) 
    print(f"[{os.getpid()}] I/O-Bound Task finished after {delay}s.")

# 3. I/O 密集型任務的協程版本
async def async_io_bound_task(delay):
    """模擬網路請求，使用 await 非阻塞等待"""
    await asyncio.sleep(delay) 
    print(f"[Asyncio] I/O-Bound Task finished after {delay}s.")

# 4. CPU 密集型任務的協程版本 (修復：使用 to_thread 將阻塞任務拋給執行緒池)
async def async_cpu_bound_task(n):
    """將 CPU 運算拋給後台執行緒池，避免阻塞事件循環"""
    # 使用 asyncio.to_thread 將同步函數 cpu_bound_task 在單獨的執行緒中運行
    # 這是 Asyncio 處理 CPU 密集型任務的標準做法。
    await asyncio.to_thread(cpu_bound_task, n) 
    print(f"[Asyncio] CPU-Bound Task finished.")

# --- 運行函數 ---

# A. 行程 (Process) 執行
def run_processes(task, *args):
    start_time = time.time()
    processes = []
    for _ in range(NUM_WORKERS):
        p = Process(target=task, args=args)
        processes.append(p)
        p.start()
    for p in processes:
        p.join()
    end_time = time.time()
    return end_time - start_time

# B. 執行緒 (Thread) 執行
def run_threads(task, *args):
    start_time = time.time()
    threads = []
    for _ in range(NUM_WORKERS):
        t = Thread(target=task, args=args)
        threads.append(t)
        t.start()
    for t in threads:
        t.join()
    end_time = time.time()
    return end_time - start_time


# C. 協程 (Asyncio) 執行
def run_asyncio(task, *args):
    start_time = time.time()
    
    # 1. 創建協程列表 (它們是協程對象)
    tasks = [task(*args) for _ in range(NUM_WORKERS)]
    
    # 2. 定義一個頂層協程，用來等待所有任務完成
    async def main_runner():
        # await 必須在 async 函數內，用於等待 Future 完成
        await asyncio.gather(*tasks)

    # 3. 運行事件循環，傳遞頂層協程 (main_runner() 返回 Coroutine Object)
    asyncio.run(main_runner())
    
    end_time = time.time()
    return end_time - start_time

# --- 主程式執行區 ---
if __name__ == "__main__":
    print(f"--- Python 併發機制測試 (N={NUM_WORKERS}, CPU_N={N_CPU}, IO_D={DELAY_IO}) ---")

    # 1. CPU 密集型任務比較 (應比較 Process vs Thread vs Asyncio)
    print("\n[--- CPU 密集型測試 (計算運算) ---]")
    
    # 【Python 1】Process：**最適用**，應最快，因繞過 GIL
    time_p_cpu = run_processes(cpu_bound_task, N_CPU)
    print(f"Process (CPU) Total Time: {time_p_cpu:.4f}s\n")
    
    # 【Python 2】Thread：**不適用**，應最慢，因受 GIL 限制
    time_t_cpu = run_threads(cpu_bound_task, N_CPU)
    print(f"Thread (CPU) Total Time: {time_t_cpu:.4f}s\n")
    
    # 【Python 3】Asyncio：**不適用**，應與 Thread 接近，因單執行緒阻塞
    time_a_cpu = run_asyncio(async_cpu_bound_task, N_CPU)
    print(f"Asyncio (CPU) Total Time: {time_a_cpu:.4f}s\n")

    # 2. I/O 密集型任務比較 (應比較 Thread vs Asyncio)
    print("\n[--- I/O 密集型測試 (網路等待) ---]")
    
    # 【Python 4】Thread：適用，因 I/O 等待時能釋放 GIL
    time_t_io = run_threads(io_bound_task, DELAY_IO)
    print(f"Thread (I/O) Total Time: {time_t_io:.4f}s\n")
    
    # 【Python 5】Asyncio：**最適用**，應最快，因切換開銷最小
    time_a_io = run_asyncio(async_io_bound_task, DELAY_IO)
    print(f"Asyncio (I/O) Total Time: {time_a_io:.4f}s\n")
    
    # Process 處理 I/O 任務通常開銷較大，一般不使用，故不列入主要比較。