Python 并发编程实战

import asyncio
import aiohttp
import time

async def download(url, session):
    """下载单个 URL"""
    try:
        async with session.get(url, timeout=10) as response:
            content = await response.read()
            return url, len(content)
    except Exception as e:
        return url, f"错误: {e}"

async def download_all(urls, max_concurrent=5):
    """并发下载多个 URL"""
    # 使用信号量限制并发数
    semaphore = asyncio.Semaphore(max_concurrent)
    
    async def bounded_download(url, session):
        async with semaphore:
            return await download(url, session)
    
    async with aiohttp.ClientSession() as session:
        tasks = [bounded_download(url, session) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results

# 使用示例
urls = [
    "https://www.python.org",
    "https://www.github.com",
    "https://www.google.com",
] * 3

async def main():
    start = time.time()
    results = await download_all(urls, max_concurrent=5)
    
    for url, result in results:
        print(f"{url}: {result}")
    
    print(f"\n总耗时: {time.time() - start:.2f}s")

# asyncio.run(main())

import asyncio
import multiprocessing
from concurrent.futures import ProcessPoolExecutor

async def cpu_bound_task(data):
    """CPU 密集型任务"""
    return sum(x * x for x in data)

def cpu_bound_sync(data):
    """同步版本用于进程池"""
    return sum(x * x for x in data)

async def io_bound_task(url):
    """IO 密集型任务"""
    await asyncio.sleep(0.1)  # 模拟网络请求
    return f"数据: {url}"

async def mixed_workflow():
    """混合工作流"""
    # IO 密集型：获取多个数据源
    urls = [f"url_{i}" for i in range(10)]
    io_tasks = [io_bound_task(url) for url in urls]
    raw_data = await asyncio.gather(*io_tasks)
    
    # CPU 密集型：处理数据（使用进程池）
    data_chunks = [[i] * 100000 for i in range(4)]
    
    loop = asyncio.get_event_loop()
    with ProcessPoolExecutor() as pool:
        # 在线程池中运行 CPU 密集型任务
        cpu_tasks = [
            loop.run_in_executor(pool, cpu_bound_sync, chunk)
            for chunk in data_chunks
        ]
        processed = await asyncio.gather(*cpu_tasks)
    
    return raw_data, processed

# asyncio.run(mixed_workflow())

import asyncio

async def producer(queue, n):
    """生产者"""
    for i in range(n):
        item = f"产品-{i}"
        await queue.put(item)
        print(f"生产: {item}")
        await asyncio.sleep(0.1)
    
    # 发送结束信号
    for _ in range(3):  # 3 个消费者
        await queue.put(None)

async def consumer(queue, name):
    """消费者"""
    while True:
        item = await queue.get()
        if item is None:
            break
        
        # 模拟处理
        await asyncio.sleep(0.2)
        print(f"消费者 {name} 处理: {item}")
        queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=10)
    
    # 创建任务
    producers = [asyncio.create_task(producer(queue, 10))]
    consumers = [
        asyncio.create_task(consumer(queue, f"C{i}"))
        for i in range(3)
    ]
    
    # 等待完成
    await asyncio.gather(*producers)
    await asyncio.gather(*consumers)
    
    print("全部完成")

asyncio.run(main())

import threading

# 错误示例：嵌套锁可能导致死锁
lock_a = threading.Lock()
lock_b = threading.Lock()

def thread1():
    with lock_a:
        print("线程1获取锁A")
        with lock_b:  # 可能死锁
            print("线程1获取锁B")

def thread2():
    with lock_b:
        print("线程2获取锁B")
        with lock_a:  # 可能死锁
            print("线程2获取锁A")

# 解决方案1：总是按相同顺序获取锁
# 解决方案2：使用 RLock（可重入锁）
# 解决方案3：使用 timeout

def safe_thread1():
    if lock_a.acquire(timeout=5):
        try:
            print("线程1获取锁A")
            if lock_b.acquire(timeout=5):
                try:
                    print("线程1获取锁B")
                finally:
                    lock_b.release()
        finally:
            lock_a.release()

import threading

# 错误：竞态条件
counter = 0

def increment():
    global counter
    # 读取 -> 修改 -> 写入 不是原子操作
    current = counter
    counter = current + 1

# 解决方案：使用锁或原子操作
lock = threading.Lock()

def safe_increment():
    global counter
    with lock:
        counter += 1

# 或使用 queue.Queue（线程安全）
# 或使用 multiprocessing.Value（进程安全）

import asyncio

# 错误：未取消任务可能导致资源泄漏
async def leaky():
    task = asyncio.create_task(asyncio.sleep(10))
    # 如果这里发生异常，任务可能仍在运行
    await asyncio.sleep(1)
    # task 仍在后台运行！

# 正确：使用 try/finally 或 async with
async def clean():
    task = asyncio.create_task(asyncio.sleep(10))
    try:
        await asyncio.sleep(1)
    finally:
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass

import asyncio

async def limited_task(sem, i):
    async with sem:
        await asyncio.sleep(1)
        return i

async def main():
    # 最多 5 个并发
    sem = asyncio.Semaphore(5)
    tasks = [limited_task(sem, i) for i in range(20)]
    results = await asyncio.gather(*tasks)
    print(f"完成 {len(results)} 个任务")

asyncio.run(main())

import asyncio

async def process_batch(items):
    """批量处理"""
    tasks = [process_item(item) for item in items]
    return await asyncio.gather(*tasks)

async def process_all(items, batch_size=10):
    """分批处理大量数据"""
    results = []
    for i in range(0, len(items), batch_size):
        batch = items[i:i + batch_size]
        batch_results = await process_batch(batch)
        results.extend(batch_results)
    return results

async def process_item(item):
    await asyncio.sleep(0.1)
    return item * 2

# 使用
# asyncio.run(process_all(range(100)))