python第十一课：并发编程 | 多任务交响乐团

news2025/3/3 22:35:57

🎯 本节目标

理解多线程/多进程/协程的应用场景
掌握threading与multiprocessing核心用法
学会使用asyncio进行异步编程
开发实战项目：高并发爬虫引擎
破解GIL锁的性能迷思

1️⃣ 并发编程三剑客

在这里插入图片描述

🎻 生活化比喻：

多线程 → 餐厅多个服务员共享厨房
进程 → 连锁餐厅各自独立分店
协程 → 一个服务员同时处理多桌点餐

2️⃣ 多线程实战：闪电下载器

import threading  
import requests  

def download(url, filename):  
    print(f"🚀 开始下载 {filename}")  
    data = requests.get(url).content  
    with open(filename, 'wb') as f:  
        f.write(data)  
    print(f"✅ {filename} 下载完成")  

# 创建线程列表  
threads = []  
urls = [  
    ('https://example.com/1.jpg', 'pic1.jpg'),  
    ('https://example.com/2.mp4', 'video.mp4')  
]  

for url, name in urls:  
    t = threading.Thread(target=download, args=(url, name))  
    threads.append(t)  
    t.start()  

# 等待所有线程完成  
for t in threads:  
    t.join()  
print("🎉 所有下载任务完成！")

3️⃣ 协程魔法：异步爬虫引擎

import asyncio  
import aiohttp  

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

async def main():  
    async with aiohttp.ClientSession() as session:  
        tasks = [  
            async_fetch(session, f"https://api.example.com/data/{i}")  
            for i in range(100)  
        ]  
        results = await asyncio.gather(*tasks)  
        print(f"📊 获取到 {len(results)} 条数据")  

# Python 3.7+ 使用 asyncio.run()  
asyncio.run(main())

4️⃣ GIL锁：性能瓶颈与突破

🔒 GIL（全局解释器锁）真相

单进程中同一时间只有一个线程执行字节码
单进程中同一时间只有一个线程执行字节码
破解方案：

# 使用多进程绕过GIL限制  
from multiprocessing import Pool  

def heavy_compute(n):  
    return sum(i*i for i in range(n))  

with Pool(4) as p:  
    results = p.map(heavy_compute, [10**6]*4)

5️⃣ 实战项目：智能并发爬虫

import concurrent.futures  
import requests  

def advanced_crawler(urls, max_workers=5):  
    """智能并发爬虫"""  
    with concurrent.futures.ThreadPoolExecutor(max_workers) as executor:  
        future_to_url = {  
            executor.submit(requests.get, url): url  
            for url in urls  
        }  
        for future in concurrent.futures.as_completed(future_to_url):  
            url = future_to_url[future]  
            try:  
                data = future.result()  
                print(f"🌐 {url} 抓取成功（长度：{len(data.text)}）")  
            except Exception as e:  
                print(f"❌ {url} 抓取失败：{str(e)}")  

# 使用示例  
url_list = [f"https://example.com/page/{i}" for i in range(50)]  
advanced_crawler(url_list, max_workers=10)

📚 知识图谱

并发编程决策树：
               ┌───────────────┐  
               │   任务类型？   │  
               └───────┬───────┘  
          ┌────────────┴────────────┐  
    ┌─────▼─────┐            ┌──────▼──────┐  
    │ I/O密集型  │            │ CPU密集型    │  
    └─────┬─────┘            └──────┬──────┘  
   ┌──────▼──────┐           ┌──────▼──────┐  
   │ 多线程/协程 │           │   多进程     │  
   └─────────────┘           └─────────────┘  

PyCharm并发调试技巧：  
 1. 线程/进程ID显示：View → Toolbar → Show Threads  
 2. 协程堆栈追踪：async堆栈模式切换  
 3. 内存/CPU监控：右键状态栏 → 勾选Memory Indicator

🛠️ 课后挑战

使用生产者-消费者模式实现多线程任务队列
将异步爬虫改造为支持断点续传
用多进程计算100个1e6大小随机数组的标准差

💡 参考答案提示:

# 生产者-消费者模型核心  
import queue  

task_queue = queue.Queue(maxsize=100)  

def producer():  
    while True:  
        item = generate_item()  
        task_queue.put(item)  

def consumer():  
    while True:  
        item = task_queue.get()  
        process_item(item)  
        task_queue.task_done()