当前位置：首页 > news >正文

Python 并发编程实战：提升程序性能

news 2026/5/7 20:10:47

Python 并发编程实战：提升程序性能

并发编程的重要性

在当今的软件开发中，并发编程变得越来越重要。随着多核处理器的普及，充分利用系统资源，提高程序性能，成为了开发者的重要任务。Python作为一种流行的编程语言，提供了多种并发编程的方式，包括多线程、多进程和异步IO。本文将介绍Python并发编程的核心概念、常用库和最佳实践。

基本概念

并发 vs 并行

并发：指在同一时间段内执行多个任务，这些任务可能是交替执行的
并行：指在同一时刻执行多个任务，这些任务是真正同时执行的

进程 vs 线程

进程：是操作系统分配资源的基本单位，每个进程有自己的内存空间
线程：是进程内的执行单元，多个线程共享进程的内存空间

同步 vs 异步

同步：指任务按顺序执行，一个任务完成后才能开始下一个任务
异步：指任务可以并行执行，不需要等待前一个任务完成

多线程编程

threading模块

import threading import time # 定义线程函数 def worker(name, delay): print(f"线程 {name} 开始执行") time.sleep(delay) print(f"线程 {name} 执行完成") # 创建线程 thread1 = threading.Thread(target=worker, args=('A', 2)) thread2 = threading.Thread(target=worker, args=('B', 3)) # 启动线程 thread1.start() thread2.start() # 等待线程完成 thread1.join() thread2.join() print("所有线程执行完成")

线程安全

import threading import time # 共享变量 counter = 0 # 锁 lock = threading.Lock() def increment(): global counter for _ in range(1000000): # 获取锁 with lock: counter += 1 # 创建线程 threads = [] for i in range(10): thread = threading.Thread(target=increment) threads.append(thread) thread.start() # 等待所有线程完成 for thread in threads: thread.join() print(f"最终计数器值: {counter}")

线程池

from concurrent.futures import ThreadPoolExecutor import time def worker(name, delay): print(f"线程 {name} 开始执行") time.sleep(delay) print(f"线程 {name} 执行完成") return f"{name} 完成" # 创建线程池 with ThreadPoolExecutor(max_workers=3) as executor: # 提交任务 future1 = executor.submit(worker, 'A', 2) future2 = executor.submit(worker, 'B', 3) future3 = executor.submit(worker, 'C', 1) # 获取结果 print(f"结果1: {future1.result()}") print(f"结果2: {future2.result()}") print(f"结果3: {future3.result()}") print("所有任务执行完成")

多进程编程

multiprocessing模块

import multiprocessing import time def worker(name, delay): print(f"进程 {name} 开始执行") time.sleep(delay) print(f"进程 {name} 执行完成") if __name__ == '__main__': # 创建进程 process1 = multiprocessing.Process(target=worker, args=('A', 2)) process2 = multiprocessing.Process(target=worker, args=('B', 3)) # 启动进程 process1.start() process2.start() # 等待进程完成 process1.join() process2.join() print("所有进程执行完成")

进程间通信

import multiprocessing def producer(queue): for i in range(5): print(f"生产: {i}") queue.put(i) queue.put(None) # 结束信号 def consumer(queue): while True: item = queue.get() if item is None: break print(f"消费: {item}") if __name__ == '__main__': # 创建队列 queue = multiprocessing.Queue() # 创建进程 producer_process = multiprocessing.Process(target=producer, args=(queue,)) consumer_process = multiprocessing.Process(target=consumer, args=(queue,)) # 启动进程 producer_process.start() consumer_process.start() # 等待进程完成 producer_process.join() consumer_process.join() print("所有进程执行完成")

进程池

from concurrent.futures import ProcessPoolExecutor import time def worker(name, delay): print(f"进程 {name} 开始执行") time.sleep(delay) print(f"进程 {name} 执行完成") return f"{name} 完成" if __name__ == '__main__': # 创建进程池 with ProcessPoolExecutor(max_workers=3) as executor: # 提交任务 future1 = executor.submit(worker, 'A', 2) future2 = executor.submit(worker, 'B', 3) future3 = executor.submit(worker, 'C', 1) # 获取结果 print(f"结果1: {future1.result()}") print(f"结果2: {future2.result()}") print(f"结果3: {future3.result()}") print("所有任务执行完成")

异步IO编程

asyncio模块

import asyncio async def worker(name, delay): print(f"任务 {name} 开始执行") await asyncio.sleep(delay) print(f"任务 {name} 执行完成") return f"{name} 完成" async def main(): # 创建任务 task1 = asyncio.create_task(worker('A', 2)) task2 = asyncio.create_task(worker('B', 3)) task3 = asyncio.create_task(worker('C', 1)) # 等待任务完成 results = await asyncio.gather(task1, task2, task3) print(f"结果: {results}") # 运行主函数 asyncio.run(main()) print("所有任务执行完成")

异步IO与网络编程

import asyncio async def handle_client(reader, writer): # 接收数据 data = await reader.read(100) message = data.decode() addr = writer.get_extra_info('peername') print(f"收到来自 {addr} 的数据: {message}") # 发送数据 response = f"你好, {addr}!" writer.write(response.encode()) await writer.drain() # 关闭连接 writer.close() await writer.wait_closed() async def main(): # 创建服务器 server = await asyncio.start_server( handle_client, 'localhost', 8888) # 获取服务器地址 addr = server.sockets[0].getsockname() print(f"服务器已启动，监听 {addr}") # 运行服务器 async with server: await server.serve_forever() # 运行主函数 asyncio.run(main())

异步IO与HTTP请求

import asyncio import aiohttp async def fetch(session, url): async with session.get(url) as response: return await response.text() async def main(): urls = [ 'https://api.github.com/users/octocat', 'https://api.github.com/users/github', 'https://api.github.com/users/python' ] async with aiohttp.ClientSession() as session: tasks = [fetch(session, url) for url in urls] results = await asyncio.gather(*tasks) for i, result in enumerate(results): print(f"URL {i+1} 响应长度: {len(result)}") # 运行主函数 asyncio.run(main()) print("所有请求完成")

并发编程的选择

适用场景

多线程：适用于IO密集型任务，如网络请求、文件IO等
多进程：适用于CPU密集型任务，如数据计算、图像处理等
异步IO：适用于IO密集型任务，尤其是需要大量并发的场景

性能对比

import time import threading import multiprocessing from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor import asyncio def cpu_intensive_task(n): """CPU密集型任务""" result = 0 for i in range(n): result += i * i return result def io_intensive_task(delay): """IO密集型任务""" time.sleep(delay) return delay # 测试多线程 start_time = time.time() with ThreadPoolExecutor(max_workers=4) as executor: results = list(executor.map(io_intensive_task, [1] * 10)) print(f"多线程IO任务耗时: {time.time() - start_time:.2f}秒") # 测试多进程 start_time = time.time() with ProcessPoolExecutor(max_workers=4) as executor: results = list(executor.map(cpu_intensive_task, [10000000] * 4)) print(f"多进程CPU任务耗时: {time.time() - start_time:.2f}秒") # 测试异步IO async def async_io_task(): tasks = [asyncio.sleep(1) for _ in range(10)] await asyncio.gather(*tasks) start_time = time.time() asyncio.run(async_io_task()) print(f"异步IO任务耗时: {time.time() - start_time:.2f}秒")

实用应用

并发下载文件

import asyncio import aiohttp import os async def download_file(session, url, save_path): """下载文件""" print(f"开始下载: {url}") async with session.get(url) as response: with open(save_path, 'wb') as f: while True: chunk = await response.content.read(1024) if not chunk: break f.write(chunk) print(f"下载完成: {save_path}") async def main(): urls = [ 'https://example.com/file1.txt', 'https://example.com/file2.txt', 'https://example.com/file3.txt', 'https://example.com/file4.txt', 'https://example.com/file5.txt' ] # 创建保存目录 os.makedirs('downloads', exist_ok=True) async with aiohttp.ClientSession() as session: tasks = [] for i, url in enumerate(urls): save_path = f'downloads/file{i+1}.txt' task = asyncio.create_task(download_file(session, url, save_path)) tasks.append(task) await asyncio.gather(*tasks) # 运行主函数 asyncio.run(main()) print("所有文件下载完成")

并发数据处理

from concurrent.futures import ProcessPoolExecutor import time def process_data(data): """处理数据""" # 模拟数据处理 time.sleep(1) return sum(data) def main(): # 生成数据 data_list = [list(range(1000000)) for _ in range(8)] # 使用进程池处理数据 start_time = time.time() with ProcessPoolExecutor(max_workers=4) as executor: results = list(executor.map(process_data, data_list)) print(f"处理完成，结果: {results}") print(f"耗时: {time.time() - start_time:.2f}秒") if __name__ == '__main__': main()

异步Web服务器

from aiohttp import web import asyncio async def handle(request): # 模拟IO操作 await asyncio.sleep(0.1) return web.Response(text="Hello, World!") async def handle_data(request): # 模拟IO操作 await asyncio.sleep(0.1) data = request.query.get('data', 'default') return web.Response(text=f"Received: {data}") async def main(): app = web.Application() app.add_routes([ web.get('/', handle), web.get('/data', handle_data) ]) runner = web.AppRunner(app) await runner.setup() site = web.TCPSite(runner, 'localhost', 8080) await site.start() print("服务器已启动，监听 http://localhost:8080") # 保持运行 await asyncio.Future() # 运行服务器 asyncio.run(main())