Python26_并发协程

Python26_并发协程

文章目录

• Python26_并发协程
• • @[toc]
  • 第一章：基础概念
  • • 1.1 什么是协程（Coroutine）？
    • 1.2 协程、线程、进程的区别
    • 1.3 核心关键字
  • 第二章：asyncio 核心机制
  • • 2.1 事件循环（Event Loop）
    • 2.2 创建和运行任务
    • 2.3 并发执行多个任务
  • 第三章：异步 I/O 与常用操作
  • • 3.1 模拟 I/O 操作
    • 3.2 异步上下文管理器
    • 3.3 异步迭代器
  • 第四章：异常处理与取消
  • • 4.1 协程中的异常
    • 4.2 任务取消（Cancellation）
  • 第五章：高级主题
  • • 5.1 在协程中运行同步代码
    • 5.2 有界信号量（Semaphore）控制并发数
    • 5.3 超时控制
  • 第六章：常见错误与调试
  • • 6.1 常见错误清单
    • 6.2 调试技巧
  • 第七章：实战模式
  • • 7.1 生产者-消费者模式（使用 Queue）
    • 7.2 Web 爬虫并发模板
  • 附录：速查表

第一章：基础概念

1.1 什么是协程（Coroutine）？

Q：协程是什么？与普通函数有什么区别？

A：协程是一种可以暂停执行并在稍后恢复的函数。与普通函数不同，协程使用async def定义，调用时不会立即执行，而是返回一个协程对象。

importasyncio# 普通函数defnormal_func():return"Hello"# 协程函数asyncdefcoro_func():return"Hello Async"# 区别演示print(normal_func())# 立即执行，输出: Helloprint(coro_func())# 不执行！输出: <coroutine object ...>print(asyncio.run(coro_func()))# 正确执行，输出: Hello Async

1.2 协程、线程、进程的区别

Q：为什么协程适合高并发 I/O？

A：协程在单线程内通过事件循环调度，遇到 I/O 操作时主动让出控制权，不阻塞其他协程。相比线程，协程切换无需内核介入，内存占用极小（约 1KB），可轻松支持数万并发连接。

1.3 核心关键字

第二章：asyncio 核心机制

2.1 事件循环（Event Loop）

Q：什么是事件循环？

A：事件循环是asyncio的核心，负责调度和执行协程。它维护一个任务队列，不断检查哪些协程可以继续执行。

importasyncioasyncdefsay_hello():print("Hello")awaitasyncio.sleep(1)# 模拟 I/O，挂起协程print("World")# 方式1：使用 asyncio.run()（推荐，Python 3.7+）asyncio.run(say_hello())# 方式2：手动管理事件循环（旧方式）loop=asyncio.get_event_loop()loop.run_until_complete(say_hello())loop.close()

2.2 创建和运行任务

Q：asyncio.create_task()和await直接调用有什么区别？

A：

• await coro()：顺序执行，等待完成才继续
• create_task()：并发执行，立即创建后台任务

importasyncioimporttimeasyncdeftask(name,delay):print(f"任务{name}开始")awaitasyncio.sleep(delay)print(f"任务{name}完成")returnf"结果{name}"asyncdefmain():start=time.time()# 方式1：顺序执行（总耗时 3秒）# result1 = await task("A", 1)# result2 = await task("B", 2)# 方式2：并发执行（总耗时 2秒）task1=asyncio.create_task(task("A",1))task2=asyncio.create_task(task("B",2))result1=awaittask1 result2=awaittask2print(f"耗时:{time.time()-start:.2f}秒")print(f"结果:{result1},{result2}")asyncio.run(main())

2.3 并发执行多个任务

Q：如何并发执行多个协程并等待全部完成？

A：使用asyncio.gather()或asyncio.wait()

importasyncioasyncdeffetch_data(url):awaitasyncio.sleep(1)# 模拟网络请求returnf"数据来自{url}"asyncdefmain():urls=["url1","url2","url3"]# 方式1：gather - 等待全部完成，返回结果列表results=awaitasyncio.gather(*[fetch_data(url)forurlinurls],return_exceptions=True# 捕获异常而不中断)print(results)# 方式2：wait - 更灵活的控制tasks=[asyncio.create_task(fetch_data(url))forurlinurls]done,pending=awaitasyncio.wait(tasks,return_when=asyncio.ALL_COMPLETED)fortaskindone:print(task.result())asyncio.run(main())

gathervswait对比：

第三章：异步 I/O 与常用操作

3.1 模拟 I/O 操作

Q：为什么time.sleep()不能在协程中使用？

A：time.sleep()会阻塞整个线程，导致事件循环无法调度其他协程。必须使用asyncio.sleep()。

importasyncioimporttimeasyncdefbad_example():print("开始")time.sleep(2)# ❌ 错误：阻塞整个事件循环！print("结束")asyncdefgood_example():print("开始")awaitasyncio.sleep(2)# ✅ 正确：挂起当前协程，让出控制权print("结束")

3.2 异步上下文管理器

Q：如何实现异步资源管理（如数据库连接）？

A：使用async with和__aenter__/__aexit__

importasyncioclassAsyncDatabase:asyncdefconnect(self):awaitasyncio.sleep(0.1)print("数据库已连接")returnselfasyncdefquery(self,sql):awaitasyncio.sleep(0.5)returnf"查询结果:{sql}"asyncdefclose(self):awaitasyncio.sleep(0.1)print("数据库已关闭")asyncdef__aenter__(self):awaitself.connect()returnselfasyncdef__aexit__(self,exc_type,exc_val,exc_tb):awaitself.close()asyncdefmain():asyncwithAsyncDatabase()asdb:result=awaitdb.query("SELECT * FROM users")print(result)asyncio.run(main())

3.3 异步迭代器

Q：如何处理异步数据流？

A：使用async for和__aiter__/__anext__

importasyncioimportrandomclassAsyncDataStream:def__init__(self,count):self.count=count self.current=0def__aiter__(self):returnselfasyncdef__anext__(self):ifself.current>=self.count:raiseStopAsyncIterationawaitasyncio.sleep(0.1)# 模拟异步获取数据self.current+=1returnf"数据块-{self.current}"asyncdefmain():stream=AsyncDataStream(3)asyncfordatainstream:print(f"接收到:{data}")asyncio.run(main())

第四章：异常处理与取消

4.1 协程中的异常

Q：协程中的异常如何捕获？

A：使用常规try/except，或在gather中设置return_exceptions=True

importasyncioasyncdefrisky_task():awaitasyncio.sleep(1)raiseValueError("出错了！")asyncdefmain():# 方式1：直接捕获try:awaitrisky_task()exceptValueErrorase:print(f"捕获异常:{e}")# 方式2：gather 中处理results=awaitasyncio.gather(risky_task(),risky_task(),return_exceptions=True# 异常会作为结果返回，不抛出)forrinresults:ifisinstance(r,Exception):print(f"任务异常:{r}")else:print(f"任务结果:{r}")asyncio.run(main())

4.2 任务取消（Cancellation）

Q：如何取消正在运行的协程？

A：使用task.cancel()，协程内部需捕获CancelledError进行清理

importasyncioasyncdeflong_running_task():try:whileTrue:print("工作中...")awaitasyncio.sleep(1)exceptasyncio.CancelledError:print("收到取消信号，正在清理...")# 执行清理操作（关闭连接、保存状态等）raise# 必须重新抛出，让取消传播asyncdefmain():task=asyncio.create_task(long_running_task())awaitasyncio.sleep(3)task.cancel()# 发送取消请求try:awaittaskexceptasyncio.CancelledError:print("任务已取消")asyncio.run(main())

第五章：高级主题

5.1 在协程中运行同步代码

Q：如何在协程中调用阻塞的同步函数？

A：使用loop.run_in_executor()将同步代码放到线程池中执行

importasyncioimporttimeimportrequests# 同步 HTTP 库defsync_http_request(url):"""同步阻塞函数"""time.sleep(2)# 模拟耗时操作returnf"同步请求结果:{url}"asyncdefmain():loop=asyncio.get_event_loop()# 在线程池中执行同步函数，不阻塞事件循环result=awaitloop.run_in_executor(None,# 默认线程池sync_http_request,"https://example.com")print(result)# 批量处理多个同步任务urls=["url1","url2","url3"]tasks=[loop.run_in_executor(None,sync_http_request,url)forurlinurls]results=awaitasyncio.gather(*tasks)print(results)asyncio.run(main())

5.2 有界信号量（Semaphore）控制并发数

Q：如何限制同时运行的协程数量？

A：使用asyncio.Semaphore

importasyncioasyncdeflimited_task(semaphore,task_id):asyncwithsemaphore:# 获取信号量，超过限制会等待print(f"任务{task_id}开始")awaitasyncio.sleep(1)print(f"任务{task_id}完成")asyncdefmain():# 最多同时运行3个协程semaphore=asyncio.Semaphore(3)tasks=[limited_task(semaphore,i)foriinrange(10)]awaitasyncio.gather(*tasks)asyncio.run(main())

5.3 超时控制

Q：如何为协程设置超时？

A：使用asyncio.wait_for()或asyncio.timeout()（Python 3.11+）

importasyncioasyncdefslow_task():awaitasyncio.sleep(10)return"完成"asyncdefmain():# 方式1：wait_for（所有版本）try:result=awaitasyncio.wait_for(slow_task(),timeout=2.0)exceptasyncio.TimeoutError:print("任务超时！")# 方式2：timeout 上下文管理器（Python 3.11+）try:asyncwithasyncio.timeout(2.0):result=awaitslow_task()exceptTimeoutError:print("任务超时！")asyncio.run(main())

第六章：常见错误与调试

6.1 常见错误清单

6.2 调试技巧

importasyncioimportlogging# 开启调试模式logging.basicConfig(level=logging.DEBUG)asyncio.run(main(),debug=True)# 查看当前所有任务asyncdefdebug_tasks():tasks=asyncio.all_tasks()fortaskintasks:print(f"任务:{task.get_name()}, 状态:{'完成'iftask.done()else'运行中'}")

第七章：实战模式

7.1 生产者-消费者模式（使用 Queue）

importasyncioasyncdefproducer(queue,n):foriinrange(n):awaitasyncio.sleep(0.5)awaitqueue.put(f"产品-{i}")print(f"生产: 产品-{i}")awaitqueue.put(None)# 结束信号asyncdefconsumer(queue,name):whileTrue:item=awaitqueue.get()ifitemisNone:breakawaitasyncio.sleep(1)# 模拟处理print(f"消费者{name}处理:{item}")asyncdefmain():queue=asyncio.Queue(maxsize=5)# 有界队列防止内存爆炸producers=[asyncio.create_task(producer(queue,5))]consumers=[asyncio.create_task(consumer(queue,i))foriinrange(2)]awaitasyncio.gather(*producers)awaitqueue.join()# 等待队列处理完成forcinconsumers:c.cancel()asyncio.run(main())

7.2 Web 爬虫并发模板

importasyncioimportaiohttp# 需要: pip install aiohttpasyncdeffetch(session,url,semaphore):asyncwithsemaphore:try:asyncwithsession.get(url,timeout=10)asresponse:returnawaitresponse.text()exceptExceptionase:returnf"Error:{e}"asyncdefmain():urls=["https://example.com"]*100semaphore=asyncio.Semaphore(10)# 限制并发10asyncwithaiohttp.ClientSession()assession:tasks=[fetch(session,url,semaphore)forurlinurls]results=awaitasyncio.gather(*tasks,return_exceptions=True)print(f"成功获取{len(results)}个页面")asyncio.run(main())

附录：速查表

# 1. 定义协程asyncdefmy_coro():awaitasyncio.sleep(1)return"done"# 2. 运行协程asyncio.run(my_coro())# 入口点，只调用一次# 3. 并发执行tasks=[asyncio.create_task(my_coro())for_inrange(10)]results=awaitasyncio.gather(*tasks)# 4. 等待第一个完成done,pending=awaitasyncio.wait(tasks,return_when=asyncio.FIRST_COMPLETED)# 5. 超时控制result=awaitasyncio.wait_for(my_coro(),timeout=5.0)# 6. 限制并发sem=asyncio.Semaphore(5)asyncwithsem:awaitmy_coro()# 7. 后台任务task=asyncio.create_task(my_coro())# ... 其他代码 ...result=awaittask# 获取结果# 8. 取消任务task.cancel()try:awaittaskexceptasyncio.CancelledError:pass