Python多线程 vs 多进程最佳实践：性能优化全攻略

import threading
import multiprocessing
import time
import requests

def fetch_url():
    requests.get("https://httpbin.org/delay/1")  # 模拟网络延迟1秒

def run_threads():
    threads = []
    start = time.time()
    for _ in range(10):
        t = threading.Thread(target=fetch_url)
        t.start()
        threads.append(t)
    for t in threads:
        t.join()
    print("线程耗时：", time.time() - start)

def run_processes():
    processes = []
    start = time.time()
    for _ in range(10):
        p = multiprocessing.Process(target=fetch_url)
        p.start()
        processes.append(p)
    for p in processes:
        p.join()
    print("进程耗时：", time.time() - start)

if __name__ == '__main__':
    print("=== I/O 密集型测试 ===")
    run_threads()
    run_processes()

线程耗时：约1.1秒
进程耗时：约2.8秒

def fib(n):
    if n <= 2:
        return 1
    return fib(n - 1) + fib(n - 2)

def run_threads_cpu():
    threads = []
    start = time.time()
    for _ in range(4):
        t = threading.Thread(target=fib, args=(35,))
        t.start()
        threads.append(t)
    for t in threads:
        t.join()
    print("线程耗时：", time.time() - start)

def run_processes_cpu():
    processes = []
    start = time.time()
    for _ in range(4):
        p = multiprocessing.Process(target=fib, args=(35,))
        p.start()
        processes.append(p)
    for p in processes:
        p.join()
    print("进程耗时：", time.time() - start)

if __name__ == '__main__':
    print("=== CPU 密集型测试 ===")
    run_threads_cpu()
    run_processes_cpu()

线程耗时：约9秒
进程耗时：约2.8秒

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

with ThreadPoolExecutor(max_workers=5) as executor:
    results = list(executor.map(fetch_url, range(10)))

with ProcessPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(fib, [35]*4))

import os
print(os.cpu_count())  # 用于设置多进程数