—— threading/multiprocessing：让 Python 飞起来
"""Python GIL（全局解释器锁）的影响：
- GIL 保证同一时刻只有一个线程执行 Python 字节码
- I/O 密集型（网络/文件/数据库）：多线程有效，GIL 在 I/O 等待时释放
- CPU 密集型（计算/加密/图像处理）：多线程无效，需要多进程绕过 GIL
选择策略：
┌──────────────┬──────────────┬──────────────┐
│ 任务类型 │ 推荐方案 │ 原因 │
├──────────────┼──────────────┼──────────────┤
│ I/O 密集型 │ threading │ GIL 自动释放 │
│ CPU 密集型 │ multiprocessing│ 绕过 GIL │
│ 混合型 │ asyncio │ 单线程并发 │
└──────────────┴──────────────┴──────────────┘
"""import threading
import time
from concurrent.futures import ThreadPoolExecutor
# 方式一：直接创建线程
def fetch_url(url, delay=1):
"""模拟网络请求"""
print(f" 开始: {url}")
time.sleep(delay) # 模拟 I/O 等待
print(f" 完成: {url}")
return f"result_{url}"
threads = []
urls = [f"url_{i}" for i in range(5)]
for url in urls:
t = threading.Thread(target=fetch_url, args=(url,))
t.start()
threads.append(t)
for t in threads:
t.join() # 等待所有线程完成
print("所有线程完成")
# 方式二：ThreadPoolExecutor（推荐）
with ThreadPoolExecutor(max_workers=3) as pool:
results = pool.map(fetch_url, urls)
print(f"结果: {list(results)}")
# 方式三：submit + as_completed（更灵活）
with ThreadPoolExecutor(max_workers=3) as pool:
futures = {pool.submit(fetch_url, url): url for url in urls}
for future in as_completed(futures):
url = futures[future]
try:
result = future.result()
print(f" {url} → {result}")
except Exception as e:
print(f" {url} 异常: {e}")
# 线程安全：Lock
counter = 0
lock = threading.Lock()
def safe_increment(n):
global counter
for _ in range(n):
with lock:
counter += 1
threads = [threading.Thread(target=safe_increment, args=(100000,)) for _ in range(5)]
for t in threads:
t.start()
for t in threads:
t.join()
print(f"最终计数: {counter} (期望: 500000)")import multiprocessing
import time
from concurrent.futures import ProcessPoolExecutor
# CPU 密集型任务
def compute_prime(n):
"""计算第n个素数"""
count = 0
num = 2
while True:
if all(num % i != 0 for i in range(2, int(num**0.5) + 1)):
count += 1
if count == n:
return num
num += 1
# 单进程
start = time.time()
results_single = [compute_prime(1000) for _ in range(4)]
single_time = time.time() - start
# 多进程
start = time.time()
with ProcessPoolExecutor(max_workers=4) as pool:
results_multi = list(pool.map(compute_prime, [1000] * 4))
multi_time = time.time() - start
print(f"单进程: {single_time:.2f}s")
print(f"多进程: {multi_time:.2f}s")
print(f"加速比: {single_time/multi_time:.1f}x")
# 进程间通信：Queue
def producer(queue):
for i in range(5):
queue.put(f"item_{i}")
print(f" 生产: item_{i}")
def consumer(queue):
while True:
item = queue.get()
if item is None:
break
print(f" 消费: {item}")
queue = multiprocessing.Queue()
p1 = multiprocessing.Process(target=producer, args=(queue,))
p2 = multiprocessing.Process(target=consumer, args=(queue,))
p1.start()
p2.start()
p1.join()
queue.put(None) # 发送结束信号
p2.join()import multiprocessing
# 共享内存
def worker_with_shared(num, shared_value, shared_array, lock):
with lock:
shared_value.value += 1
shared_array[num] = num * 10
if __name__ == "__main__":
lock = multiprocessing.Lock()
shared_val = multiprocessing.Value("i", 0) # 整数
shared_arr = multiprocessing.Array("d", [0]*5) # 双精度数组
processes = []
for i in range(5):
p = multiprocessing.Process(
target=worker_with_shared,
args=(i, shared_val, shared_arr, lock)
)
processes.append(p)
p.start()
for p in processes:
p.join()
print(f"共享值: {shared_val.value}")
print(f"共享数组: {list(shared_arr)}")#!/usr/bin/env python3
"""第28课 并发编程验证"""
import threading
import multiprocessing
import time
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
def test_threading():
results = []
def worker(n):
time.sleep(0.01)
results.append(n)
threads = [threading.Thread(target=worker, args=(i,)) for i in range(10)]
for t in threads:
t.start()
for t in threads:
t.join()
assert len(results) == 10
print("✅ threading测试通过")
def test_thread_pool():
def square(n):
return n * n
with ThreadPoolExecutor(max_workers=4) as pool:
results = list(pool.map(square, range(10)))
assert results == [i*i for i in range(10)]
print("✅ ThreadPoolExecutor测试通过")
def test_lock():
counter = 0
lock = threading.Lock()
def increment():
nonlocal counter
for _ in range(1000):
with lock:
counter += 1
threads = [threading.Thread(target=increment) for _ in range(5)]
for t in threads:
t.start()
for t in threads:
t.join()
assert counter == 5000
print("✅ Lock线程安全测试通过")
def test_process_pool():
def cube(n):
return n ** 3
with ProcessPoolExecutor(max_workers=2) as pool:
results = list(pool.map(cube, range(5)))
assert results == [0, 1, 8, 27, 64]
print("✅ ProcessPoolExecutor测试通过")
if __name__ == "__main__":
test_threading()
test_thread_pool()
test_lock()
test_process_pool()
print("\n🎉 第28课全部验证通过！")import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
from queue import Queue
class WorkerPool:
"""生产级线程池"""
def __init__(self, max_workers=5):
self.max_workers = max_workers
self.results = Queue()
self.errors = []
def submit_tasks(self, task_func, items):
with ThreadPoolExecutor(max_workers=self.max_workers) as pool:
futures = {pool.submit(task_func, item): item for item in items}
for future in as_completed(futures):
try:
result = future.result()
self.results.put(result)
except Exception as e:
self.errors.append({"item": futures[future], "error": str(e)})
def get_results(self):
results = []
while not self.results.empty():
results.append(self.results.get())
return results
# 使用
import time
def process_url(url):
time.sleep(0.01)
return f"result_{url}"
pool = WorkerPool(max_workers=3)
pool.submit_tasks(process_url, [f"url_{i}" for i in range(10)])
print(f"结果: {len(pool.get_results())} 个, 错误: {len(pool.errors)} 个")import multiprocessing
from concurrent.futures import ProcessPoolExecutor
import math
def is_prime(n):
if n < 2: return False
if n < 4: return True
if n % 2 == 0: return False
for i in range(3, int(math.sqrt(n)) + 1, 2):
if n % i == 0: return False
return True
def count_primes_range(start, end):
return sum(1 for n in range(start, end) if is_prime(n))
def parallel_prime_count(max_num, num_workers=None):
num_workers = num_workers or multiprocessing.cpu_count()
chunk_size = max_num // num_workers
ranges = [(i * chunk_size, (i+1) * chunk_size if i < num_workers-1 else max_num)
for i in range(num_workers)]
with ProcessPoolExecutor(max_workers=num_workers) as pool:
results = list(pool.map(count_primes_range, *zip(*ranges)))
total = sum(results)
print(f"0-{max_num} 范围内有 {total} 个素数 ({num_workers}进程)")
return totalimport threading
import multiprocessing
import asyncio
import time
import concurrent.futures
# 场景1：I/O 密集（网络请求模拟）
def io_task():
time.sleep(0.01)
return "done"
# 串行
def serial_io(n):
start = time.time()
for _ in range(n):
io_task()
return time.time() - start
# 多线程
def threaded_io(n, workers=10):
start = time.time()
with concurrent.futures.ThreadPoolExecutor(workers) as pool:
list(pool.map(lambda _: io_task(), range(n)))
return time.time() - start
# 场景2：CPU 密集（计算）
def cpu_task(n):
return sum(i * i for i in range(n))
# 串行
def serial_cpu(tasks):
start = time.time()
for n in tasks:
cpu_task(n)
return time.time() - start
# 多进程
def parallel_cpu(tasks, workers=4):
start = time.time()
with concurrent.futures.ProcessPoolExecutor(workers) as pool:
list(pool.map(cpu_task, tasks))
return time.time() - start
# 对比
n = 100
t_serial = serial_io(n)
t_threaded = threaded_io(n)
print(f"I/O 密集 ({n}个任务):")
print(f" 串行: {t_serial:.2f}s")
print(f" 多线程: {t_threaded:.2f}s")
print(f" 加速: {t_serial/t_threaded:.1f}x")
tasks = [500000] * 4
t_serial_cpu = serial_cpu(tasks)
t_parallel_cpu = parallel_cpu(tasks)
print(f"
CPU 密集 ({len(tasks)}个任务):")
print(f" 串行: {t_serial_cpu:.2f}s")
print(f" 多进程: {t_parallel_cpu:.2f}s")
print(f" 加速: {t_serial_cpu/t_parallel_cpu:.1f}x")import threading
from queue import Queue, LifoQueue, PriorityQueue
from collections import defaultdict
class ThreadSafeDict:
"""线程安全字典"""
def __init__(self):
self._dict = {}
self._lock = threading.Lock()
def get(self, key, default=None):
with self._lock:
return self._dict.get(key, default)
def set(self, key, value):
with self._lock:
self._dict[key] = value
def update(self, **kwargs):
with self._lock:
self._dict.update(kwargs)
def items(self):
with self._lock:
return list(self._dict.items())
class ThreadSafeCounter:
"""线程安全计数器"""
def __init__(self):
self._counts = defaultdict(int)
self._lock = threading.Lock()
def increment(self, key="default"):
with self._lock:
self._counts[key] += 1
def get(self, key="default"):
with self._lock:
return self._counts[key]
def top(self, n=10):
with self._lock:
return sorted(self._counts.items(), key=lambda x: x[1], reverse=True)[:n]
# 队列选择指南
# Queue (FIFO) → 先进先出，最常用
# LifoQueue (栈) → 后进先出
# PriorityQueue → 优先级队列，元素为 (priority, item)
# SimpleQueue → 无任务的简化队列（Python 3.7+）