1. Python多任务编程概述在Python开发中当我们需要同时处理多个任务时多任务编程就成为了必备技能。Python提供了三种主要的并发编程方式进程(Process)、线程(Thread)和协程(Coroutine)。这三种方式各有特点适用于不同的场景。进程是操作系统资源分配的基本单位每个进程都有独立的内存空间进程间通信需要通过IPC机制。线程是进程的执行单元同一进程内的多个线程共享进程的内存空间。协程则是更轻量级的用户级线程由程序自身控制调度。Python中由于GIL(全局解释器锁)的存在多线程在CPU密集型任务上表现不佳但对于IO密集型任务多线程和多协程都能很好地发挥作用。而多进程则可以充分利用多核CPU的优势。2. 进程编程详解2.1 进程的基本概念进程是操作系统进行资源分配和调度的基本单位。每个进程都有自己独立的内存空间这使得进程间的数据隔离性很好但同时也带来了较高的资源开销。在Python中我们可以使用multiprocessing模块来创建和管理进程。这个模块提供了与threading模块类似的API使得从线程切换到进程变得相对容易。import multiprocessing def worker(num): print(fWorker: {num}) if __name__ __main__: processes [] for i in range(5): p multiprocessing.Process(targetworker, args(i,)) processes.append(p) p.start() for p in processes: p.join()2.2 进程间通信由于进程间内存隔离进程间通信(IPC)需要通过特殊机制实现。Python的multiprocessing模块提供了多种IPC方式队列(Queue)进程安全的先进先出数据结构管道(Pipe)一对连接对象用于双向通信共享内存(Value/Array)通过共享内存实现数据共享Manager更高级的共享对象管理from multiprocessing import Process, Queue def producer(q): for i in range(5): q.put(i) print(fProduced {i}) def consumer(q): while True: item q.get() if item is None: # 哨兵值 break print(fConsumed {item}) if __name__ __main__: q Queue() p1 Process(targetproducer, args(q,)) p2 Process(targetconsumer, args(q,)) p1.start() p2.start() p1.join() q.put(None) # 发送结束信号 p2.join()2.3 进程池创建进程的开销较大对于大量短期任务使用进程池(Process Pool)是更好的选择。Python提供了Pool类来管理进程池。from multiprocessing import Pool import time def task(n): print(fProcessing {n}) time.sleep(2) return n * n if __name__ __main__: with Pool(processes4) as pool: results pool.map(task, range(10)) print(results)注意在Windows系统上使用多进程时必须将主程序代码放在if __name__ __main__:块中这是为了避免子进程无限递归创建新进程。3. 线程编程详解3.1 线程的基本概念线程是进程内的执行单元是CPU调度的基本单位。同一进程内的线程共享进程的内存空间这使得线程间通信更加方便但也带来了数据同步的问题。Python通过threading模块提供线程支持。创建线程的开销比进程小很多适合IO密集型任务。import threading import time def worker(num): print(fThread {num} started) time.sleep(2) print(fThread {num} finished) threads [] for i in range(3): t threading.Thread(targetworker, args(i,)) threads.append(t) t.start() for t in threads: t.join()3.2 线程同步由于线程共享内存当多个线程访问共享资源时需要使用同步机制来避免竞态条件。Python提供了多种同步原语锁(Lock)最基本的同步机制可重入锁(RLock)允许同一线程多次获取的锁条件变量(Condition)用于线程间通信信号量(Semaphore)限制资源访问数量事件(Event)线程间通知机制import threading counter 0 lock threading.Lock() def increment(): global counter for _ in range(100000): with lock: counter 1 threads [] for _ in range(5): t threading.Thread(targetincrement) threads.append(t) t.start() for t in threads: t.join() print(fFinal counter value: {counter})3.3 线程池与进程池类似Python也提供了线程池的实现可以通过concurrent.futures模块中的ThreadPoolExecutor来使用。from concurrent.futures import ThreadPoolExecutor import time def task(n): print(fProcessing {n}) time.sleep(2) return n * n with ThreadPoolExecutor(max_workers4) as executor: results list(executor.map(task, range(5))) print(results)实际经验在IO密集型任务中线程池的大小可以设置得较大(如CPU核心数的5-10倍)而对于CPU密集型任务由于GIL的存在使用多线程反而可能降低性能。4. 协程编程详解4.1 协程的基本概念协程是一种用户态的轻量级线程由程序自身控制调度。协程的切换不涉及操作系统内核因此开销极小。一个线程可以包含多个协程Python中的协程通过asyncio模块实现。协程特别适合处理大量IO密集型任务如网络请求、文件操作等。import asyncio async def hello(name): print(fHello, {name}) await asyncio.sleep(1) print(fGoodbye, {name}) async def main(): await asyncio.gather( hello(Alice), hello(Bob), hello(Charlie) ) asyncio.run(main())4.2 协程与异步IOPython的协程通常与异步IO结合使用。asyncio模块提供了事件循环机制可以高效地管理大量网络连接或其他IO操作。import asyncio import aiohttp async def fetch_url(url): async with aiohttp.ClientSession() as session: async with session.get(url) as response: return await response.text() async def main(): urls [ http://example.com, http://example.org, http://example.net ] tasks [fetch_url(url) for url in urls] results await asyncio.gather(*tasks) for url, content in zip(urls, results): print(f{url}: {len(content)} bytes) asyncio.run(main())4.3 协程的实际应用在实际开发中协程常用于以下场景高性能网络服务器爬虫程序微服务架构中的服务调用任何需要高并发的IO密集型应用import asyncio from datetime import datetime async def task(name, delay): print(fTask {name} started at {datetime.now()}) await asyncio.sleep(delay) print(fTask {name} finished at {datetime.now()}) return fResult from {name} async def main(): # 并行执行多个任务 results await asyncio.gather( task(A, 2), task(B, 1), task(C, 3) ) print(results) asyncio.run(main())5. 多任务编程实战对比5.1 性能对比测试让我们通过一个实际的IO密集型任务来比较三种方式的性能差异。我们将模拟一个需要等待1秒的网络请求执行10次。import time import threading import multiprocessing import asyncio def io_task(): time.sleep(1) # 模拟IO操作 return result # 多线程版本 def threading_version(): threads [] for _ in range(10): t threading.Thread(targetio_task) t.start() threads.append(t) for t in threads: t.join() # 多进程版本 def processing_version(): processes [] for _ in range(10): p multiprocessing.Process(targetio_task) p.start() processes.append(p) for p in processes: p.join() # 协程版本 async def async_task(): await asyncio.sleep(1) return result async def async_version(): tasks [async_task() for _ in range(10)] await asyncio.gather(*tasks) # 测试函数 def run_test(name, func): start time.time() if name Async: asyncio.run(func()) else: func() duration time.time() - start print(f{name:10}: {duration:.2f} seconds) if __name__ __main__: print(Performance comparison (10 IO tasks):) run_test(Threading, threading_version) run_test(Processing, processing_version) run_test(Async, async_version)5.2 选择正确的并发模型在实际项目中选择哪种并发模型取决于具体需求CPU密集型任务优先考虑多进程可以充分利用多核CPUIO密集型任务如果并发量不大(几百以内)使用多线程如果并发量很大(数千以上)使用协程需要与C/C扩展交互多线程可能更合适需要跨机器扩展考虑多进程或多机分布式方案5.3 常见问题与解决方案问题1多线程中的GIL限制解决方案对于CPU密集型任务可以使用多进程或将关键部分用C扩展实现问题2进程间通信复杂解决方案根据数据量和频率选择合适的IPC机制小数据用Queue/pipe大数据考虑共享内存问题3协程调试困难解决方案使用asyncio.debug模式合理使用日志记录协程执行流程问题4资源竞争导致死锁解决方案遵循锁的获取顺序一致性原则使用超时机制或考虑使用无锁数据结构6. 高级主题与最佳实践6.1 混合使用多种并发模型在实际复杂应用中我们常常需要混合使用多种并发模型。例如可以使用多进程处理CPU密集型任务每个进程内部使用多线程或协程处理IO操作。import concurrent.futures import asyncio def cpu_intensive_task(data): # 模拟CPU密集型计算 return sum(i*i for i in range(data)) async def io_intensive_task(url): # 模拟IO操作 await asyncio.sleep(0.1) return fProcessed {url} async def mixed_approach(): # 使用进程池处理CPU密集型任务 with concurrent.futures.ProcessPoolExecutor() as process_pool: # 提交CPU密集型任务 cpu_results await asyncio.get_event_loop().run_in_executor( process_pool, cpu_intensive_task, 1000000) # 同时使用协程处理IO密集型任务 io_tasks [io_intensive_task(furl_{i}) for i in range(10)] io_results await asyncio.gather(*io_tasks) return cpu_results, io_results results asyncio.run(mixed_approach()) print(results)6.2 错误处理与容错机制在并发编程中良好的错误处理机制至关重要。以下是一些最佳实践为每个任务添加超时机制记录详细的错误日志实现重试逻辑使用上下文管理器确保资源释放import asyncio from contextlib import asynccontextmanager asynccontextmanager async def timeout_manager(task, timeout): try: yield await asyncio.wait_for(task, timeouttimeout) except asyncio.TimeoutError: print(fTask timed out after {timeout} seconds) except Exception as e: print(fTask failed with error: {str(e)}) finally: print(Cleanup resources) async def risky_operation(): await asyncio.sleep(2) # 模拟可能失败的操作 if 1 1: # 改为False可以测试成功情况 raise ValueError(Something went wrong) return Success async def main(): async with timeout_manager(risky_operation(), timeout1.5) as result: if result: print(fOperation succeeded: {result}) asyncio.run(main())6.3 性能优化技巧批量处理将小任务批量处理减少上下文切换连接池对于网络/数据库操作使用连接池缓冲区适当使用缓冲区减少IO操作次数惰性加载延迟初始化资源选择合适的并发数根据资源情况调整并发度import asyncio import aiohttp from aiohttp import TCPConnector async def optimized_fetch(urls, batch_size5): connector TCPConnector(limitbatch_size) # 限制并发连接数 async with aiohttp.ClientSession(connectorconnector) as session: tasks [] for url in urls: task asyncio.create_task( session.get(url, timeoutaiohttp.ClientTimeout(total10)) ) tasks.append(task) results [] for future in asyncio.as_completed(tasks): try: response await future results.append(await response.text()) except Exception as e: print(fError fetching URL: {str(e)}) return results urls [http://example.com] * 20 results asyncio.run(optimized_fetch(urls)) print(fFetched {len(results)} pages)7. 实战项目并发Web爬虫让我们综合运用所学知识实现一个并发的Web爬虫支持以下功能并发抓取多个页面解析页面内容存储结果错误处理和重试机制import asyncio import aiohttp from bs4 import BeautifulSoup from urllib.parse import urljoin import json from typing import Dict, List, Optional class AsyncCrawler: def __init__(self, base_url: str, max_concurrency: int 10): self.base_url base_url self.visited: Dict[str, bool] {} self.results: List[Dict] [] self.max_concurrency max_concurrency self.session: Optional[aiohttp.ClientSession] None async def fetch(self, url: str) - Optional[str]: try: async with self.session.get(url, timeoutaiohttp.ClientTimeout(total10)) as response: if response.status 200: return await response.text() return None except Exception as e: print(fError fetching {url}: {str(e)}) return None def parse(self, html: str, url: str) - Dict: soup BeautifulSoup(html, html.parser) title soup.title.string if soup.title else No title links [urljoin(url, a[href]) for a in soup.find_all(a, hrefTrue)] return { url: url, title: title, links: links } async def process_page(self, url: str): if url in self.visited: return self.visited[url] True html await self.fetch(url) if not html: return data self.parse(html, url) self.results.append(data) # 只处理同域的链接 for link in data[links]: if link.startswith(self.base_url): await self.process_page(link) async def crawl(self, start_url: str): connector aiohttp.TCPConnector(limitself.max_concurrency) self.session aiohttp.ClientSession(connectorconnector) try: await self.process_page(start_url) finally: await self.session.close() def save_results(self, filename: str): with open(filename, w, encodingutf-8) as f: json.dump(self.results, f, ensure_asciiFalse, indent2) async def main(): crawler AsyncCrawler(base_urlhttp://example.com, max_concurrency5) await crawler.crawl(http://example.com) crawler.save_results(crawl_results.json) print(fCrawled {len(crawler.results)} pages) if __name__ __main__: asyncio.run(main())这个爬虫使用了协程实现高并发抓取通过限制并发连接数避免对目标服务器造成过大压力并实现了基本的错误处理和结果存储功能。