并发编程练习答案

threading-basics

练习 1：多线程下载队列

要求：实现一个下载队列，3个工作线程并发处理URL列表，使用Queue传递任务。

答案：

import threading
import queue
import time

def download_worker(task_queue, result_queue):
    """工作线程函数"""
    while True:
        url = task_queue.get()
        if url is None:  # 停止信号
            task_queue.task_done()
            break
        
        # 模拟下载
        print(f"{threading.current_thread().name} 下载 {url}")
        time.sleep(1)
        result_queue.put(f"{url} 完成")
        
        task_queue.task_done()

# 创建队列
task_queue = queue.Queue()
result_queue = queue.Queue()

# 启动工作线程
workers = []
for i in range(3):
    t = threading.Thread(target=download_worker, args=(task_queue, result_queue))
    t.start()
    workers.append(t)

# 添加任务
urls = [f"http://example.com/file{i}.dat" for i in range(10)]
for url in urls:
    task_queue.put(url)

# 等待所有任务完成
task_queue.join()

# 停止工作线程
for _ in workers:
    task_queue.put(None)

for t in workers:
    t.join()

# 收集结果
while not result_queue.empty():
    print(result_queue.get())

关键点：

• 使用 task_done() 和 join() 等待所有任务完成
• 哨兵值 None 通知线程停止
• Queue 是线程安全的，无需额外加锁

练习 2：生产者-消费者模式

要求：2个生产者线程生成数据，3个消费者线程处理数据，使用Condition同步。

答案：

import threading
import time
import random

condition = threading.Condition()
buffer = []
MAX_SIZE = 5

def producer(name):
    for i in range(5):
        item = f"{name}-{i}"
        time.sleep(random.uniform(0.1, 0.5))
        
        with condition:
            while len(buffer) >= MAX_SIZE:
                print(f"{name} 等待（缓冲区满）")
                condition.wait()
            
            buffer.append(item)
            print(f"{name} 生产: {item}")
            condition.notify_all()

def consumer(name):
    while True:
        with condition:
            while not buffer:
                condition.wait()
            
            item = buffer.pop(0)
            print(f"{name} 消费: {item}")
            condition.notify_all()
            
            if item.endswith("-4"):  # 最后一个
                break

# 启动线程
producers = [threading.Thread(target=producer, args=(f"P{i}",)) for i in range(2)]
consumers = [threading.Thread(target=consumer, args=(f"C{i}",)) for i in range(3)]

for t in producers + consumers:
    t.start()

for t in producers + consumers:
    t.join()

关键点：

• while not buffer 循环检查条件（防止虚假唤醒）
• notify_all() 唤醒所有等待的线程
• 缓冲区满时生产者等待，空时消费者等待

练习 3：线程安全的计数器

要求：实现一个线程安全的计数器类，支持增加、减少和获取值。

答案：

import threading

class SafeCounter:
    def __init__(self):
        self._value = 0
        self._lock = threading.Lock()
    
    def increment(self, delta=1):
        with self._lock:
            self._value += delta
    
    def decrement(self, delta=1):
        with self._lock:
            self._value -= delta
    
    def get_value(self):
        with self._lock:
            return self._value

# 测试
counter = SafeCounter()

def worker():
    for _ in range(10000):
        counter.increment()

threads = [threading.Thread(target=worker) for _ in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(f"最终值: {counter.get_value()}")  # 应该是 50000

关键点：

• 所有访问共享数据的方法都要加锁
• 使用 with 语句确保锁一定会释放

---

multiprocessing

练习 1：并行素数计算

要求：使用进程池计算1到1000000之间的所有素数。

答案：

import multiprocessing
import math

def is_prime(n):
    if n < 2:
        return False
    for i in range(2, int(math.sqrt(n)) + 1):
        if n % i == 0:
            return False
    return True

def count_primes_in_range(start, end):
    """计算范围内的素数数量"""
    count = 0
    for n in range(start, end):
        if is_prime(n):
            count += 1
    return count

if __name__ == '__main__':
    # 将范围分成多个块
    total = 1000000
    num_processes = multiprocessing.cpu_count()
    chunk_size = total // num_processes
    
    ranges = [(i * chunk_size, (i + 1) * chunk_size) 
              for i in range(num_processes)]
    ranges[-1] = (ranges[-1][0], total)  # 调整最后一个范围
    
    with multiprocessing.Pool(processes=num_processes) as pool:
        results = pool.starmap(count_primes_in_range, ranges)
    
    total_primes = sum(results)
    print(f"1到{total}之间有 {total_primes} 个素数")

关键点：

• starmap 展开元组参数
• 根据 CPU 核心数分配任务
• 使用 if __name__ == '__main__': 保护

练习 2：进程间共享状态

要求：多个进程更新共享计数器，使用Lock确保正确性。

答案：

import multiprocessing

def increment_shared(shared_value, lock, count):
    for _ in range(count):
        with lock:
            shared_value.value += 1

if __name__ == '__main__':
    shared_value = multiprocessing.Value('i', 0)
    lock = multiprocessing.Lock()
    
    processes = []
    for _ in range(4):
        p = multiprocessing.Process(
            target=increment_shared,
            args=(shared_value, lock, 25000)
        )
        p.start()
        processes.append(p)
    
    for p in processes:
        p.join()
    
    print(f"最终值: {shared_value.value}")  # 应该是 100000

关键点：

• Value('i', 0) 创建共享整数
• 必须使用Lock保护共享内存访问
• 进程间Lock与线程Lock用法相同

---

asyncio-basics

练习 1：并发HTTP请求

要求：使用asyncio并发获取多个URL，统计总耗时。

答案（需要 aiohttp）：

import asyncio
import aiohttp
import time

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = [
        "https://www.python.org",
        "https://docs.python.org",
        "https://pypi.org",
    ]
    
    start = time.time()
    
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
    
    elapsed = time.time() - start
    
    for url, content in zip(urls, results):
        print(f"{url}: {len(content)} 字节")
    
    print(f"总耗时: {elapsed:.2f}秒")

asyncio.run(main())

关键点：

• async with 管理会话生命周期
• gather 并发执行所有请求
• 共享同一个 ClientSession 提高性能

练习 2：异步生产者-消费者

要求：使用asyncio.Queue实现异步生产者-消费者模式。

答案：

import asyncio
import random

async def producer(queue, name):
    for i in range(5):
        await asyncio.sleep(random.uniform(0.1, 0.5))
        item = f"{name}-{i}"
        await queue.put(item)
        print(f"{name} 生产: {item}")

async def consumer(queue, name):
    while True:
        item = await queue.get()
        print(f"{name} 消费: {item}")
        await asyncio.sleep(random.uniform(0.2, 0.6))
        queue.task_done()

async def main():
    queue = asyncio.Queue()
    
    # 创建生产者和消费者
    producers = [
        asyncio.create_task(producer(queue, f"P{i}"))
        for i in range(2)
    ]
    consumers = [
        asyncio.create_task(consumer(queue, f"C{i}"))
        for i in range(3)
    ]
    
    # 等待生产者完成
    await asyncio.gather(*producers)
    
    # 等待队列清空
    await queue.join()
    
    # 取消消费者
    for c in consumers:
        c.cancel()

asyncio.run(main())

关键点：

• asyncio.Queue 是异步队列
• 使用 await 而不是阻塞
• 生产完成后取消消费者任务

---

concurrent-futures

练习 1：比较执行器性能

要求：对比ThreadPoolExecutor和ProcessPoolExecutor处理CPU密集型任务的性能。

答案：

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time

def cpu_task(n):
    """CPU密集型任务"""
    total = 0
    for i in range(n):
        total += i ** 2
    return total

def benchmark(executor_class, workers, tasks):
    start = time.time()
    with executor_class(max_workers=workers) as executor:
        results = list(executor.map(cpu_task, tasks))
    return time.time() - start

if __name__ == '__main__':
    tasks = [1000000] * 8
    
    print("=== CPU密集型任务 ===")
    thread_time = benchmark(ThreadPoolExecutor, 4, tasks)
    print(f"ThreadPoolExecutor: {thread_time:.2f}秒")
    
    process_time = benchmark(ProcessPoolExecutor, 4, tasks)
    print(f"ProcessPoolExecutor: {process_time:.2f}秒")
    
    print(f"加速比: {thread_time / process_time:.2f}x")

关键点：

• ProcessPoolExecutor 在CPU密集型任务上显著更快
• ThreadPoolExecutor 受GIL限制，无真正并行

练习 2：带超时的批量任务

要求：提交多个任务，对每个任务设置超时，收集成功和失败的结果。

答案：

from concurrent.futures import ThreadPoolExecutor, TimeoutError
import time
import random

def task(name, delay):
    time.sleep(delay)
    return f"{name} 完成"

def process_with_timeout(executor, tasks, timeout):
    futures = {executor.submit(task, name, delay): name 
               for name, delay in tasks}
    
    results = {'success': [], 'timeout': [], 'error': []}
    
    for future in futures:
        name = futures[future]
        try:
            result = future.result(timeout=timeout)
            results['success'].append((name, result))
        except TimeoutError:
            results['timeout'].append(name)
            future.cancel()
        except Exception as e:
            results['error'].append((name, str(e)))
    
    return results

if __name__ == '__main__':
    tasks = [(f"任务{i}", random.uniform(0.5, 3)) for i in range(10)]
    
    with ThreadPoolExecutor(max_workers=4) as executor:
        results = process_with_timeout(executor, tasks, timeout=2)
    
    print(f"成功: {len(results['success'])}")
    print(f"超时: {len(results['timeout'])}")
    print(f"失败: {len(results['error'])}")

关键点：

• future.result(timeout=N) 设置单个任务超时
• 分别处理成功、超时和异常情况
• future.cancel() 尝试取消超时任务