import concurrent.futures
import time
import random

def simulate_task(seconds):
    print(f"任务开始，预计耗时 {seconds} 秒...")
    time.sleep(seconds)  # 模拟阻塞操作
    return f"任务完成，耗时 {seconds} 秒"

# 使用线程池执行任务
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    # 提交5个任务到线程池
    futures = [executor.submit(simulate_task, random.randint(1, 5)) for _ in range(5)]
    
    # 获取任务结果
    for future in concurrent.futures.as_completed(futures):
        print(future.result())

print("所有任务已完成！")
                    

import asyncio
import aiohttp

async def fetch_url(url):
    print(f"开始获取: {url}")
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            content = await response.text()
            print(f"{url} 获取完成，长度: {len(content)}")
            return content

async def main():
    urls = [
        'https://www.python.org',
        'https://www.google.com',
        'https://www.github.com'
    ]
    
    tasks = [fetch_url(url) for url in urls]
    await asyncio.gather(*tasks)

# 运行异步程序
asyncio.run(main())
                    

import threading
import queue
import time
import random

# 创建任务队列
task_queue = queue.Queue(maxsize=10)

def producer():
    """生产者线程，生成任务"""
    for i in range(1, 11):
        task = f"任务 {i}"
        task_queue.put(task)
        print(f"已添加: {task}")
        time.sleep(random.uniform(0.1, 0.5))  # 随机延迟
    task_queue.put(None)  # 结束信号

def consumer():
    """消费者线程，处理任务"""
    while True:
        task = task_queue.get()
        if task is None:  # 遇到结束信号
            task_queue.put(None)  # 通知其他消费者
            break
        print(f"处理中: {task}")
        time.sleep(random.uniform(0.5, 1.5))  # 模拟处理时间
        task_queue.task_done()

# 创建并启动线程
prod_thread = threading.Thread(target=producer)
cons_thread1 = threading.Thread(target=consumer)
cons_thread2 = threading.Thread(target=consumer)

prod_thread.start()
cons_thread1.start()
cons_thread2.start()

# 等待所有任务完成
prod_thread.join()
task_queue.join()  # 等待所有任务被处理
cons_thread1.join()
cons_thread2.join()

print("所有任务处理完毕")
                    

4 使用多进程处理CPU密集型任务

对于CPU密集型任务，使用多进程绕过GIL限制，充分利用多核CPU。

from multiprocessing import Pool

def cpu_intensive(n):
    # 模拟CPU密集型任务
    return sum(i * i for i in range(n))

if __name__ == '__main__':
    with Pool(4) as p:  # 使用4个进程
        results = p.map(cpu_intensive, [10000000, 20000000, 30000000])
        print(results)
                        

5 使用非阻塞I/O和超时机制

设置I/O操作超时，使用非阻塞I/O库，防止线程无限期阻塞。

import socket

# 创建非阻塞socket
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setblocking(False)  # 设置为非阻塞模式

try:
    # 尝试连接，设置超时5秒
    sock.connect(('www.example.com', 80))
except BlockingIOError:
    # 非阻塞模式下正常出现的异常
    pass

# 等待连接完成或超时
timeout = 5
start_time = time.time()
while True:
    try:
        # 检查连接是否就绪
        sock.send(b'GET / HTTP/1.0\r\n\r\n')
        break
    except OSError as e:
        if time.time() - start_time > timeout:
            raise TimeoutError("连接超时")
        time.sleep(0.1)
                        

任务类型决策树

• I/O密集型 → 使用asyncio或线程池
• CPU密集型 → 使用多进程
• 混合型任务 → 组合使用进程池和线程池
• 高并发网络 → 异步框架如FastAPI

避免常见陷阱

• 避免在多线程中使用可变全局状态
• 使用线程安全的数据结构
• 合理设置线程/进程池大小
• 使用上下文管理器确保资源释放
• 添加超时机制防止永久阻塞

调试与监控工具

• threading.enumerate() - 查看活动线程
• logging模块 - 线程安全日志记录
• cProfile - 性能分析
• 第三方库：objgraph, memory_profiler
• 可视化工具：PyCharm调试器