Py网络基础编程 7

ComputerNetworks第六章异步编程与协程提升程序性能的关键技术《计算机网络》课程Contents本章目录6.1异步编程概述同步与异步编程模型、异步编程的优势与应用场景6.2协程基础协程的概念、async/await语法、协程执行流程6.3asyncio模块详解事件循环、Task与Future对象、常用API函数6.4异步I/O操作StreamReader/StreamWriter、TCP服务器、Protocol与Transport6.5Web应用开发aiohttp库、HTTP客户端与服务器、WebSocket编程学习目标掌握异步编程核心概念，构建高性能网络应用6.1Chapter6.1异步编程概述理解同步与异步编程的核心差异SyncvsAsync同步编程vs异步编程同步编程同步（Synchronous）：程序按照代码书写顺序依次执行，每个操作必须等待前一个操作完成后才能开始。执行特点：•阻塞式执行，一个任务未完成，后续任务只能等待•代码逻辑简单清晰，易于理解和调试•适合CPU密集型任务（如计算、数据处理）异步编程异步（Asynchronous）：程序在等待某个操作完成时，可以继续执行其他操作，无需阻塞等待。执行特点：•非阻塞式执行，任务等待期间可执行其他任务•代码逻辑相对复杂，需要处理回调或协程•适合I/O密集型任务（如网络请求、文件读写）执行模型对比同步执行模型任务1任务2任务3顺序执行，前一个任务阻塞时整个程序等待异步执行模型任务1启动任务2执行任务3执行任务1完成任务等待期间可切换执行其他任务，提高执行效率关键差异总结执行方式：同步阻塞vs异步非阻塞资源利用：低vs高编程复杂度：简单vs相对复杂适用场景：CPU密集型vsI/O密集型WhyAsynchronous为什么需要异步编程解决I/O阻塞问题在传统的同步编程中，当程序执行I/O操作（如网络请求、文件读写、数据库查询）时，线程会被阻塞，直到操作完成才能继续执行。这种阻塞会导致CPU资源浪费，程序响应变慢。异步编程通过事件驱动机制，在I/O等待期间释放CPU，让程序可以继续处理其他任务，从而最大化资源利用率。提高并发性能异步编程能够在单线程内处理大量并发连接，相比多线程/多进程模型，具有更低的内存占用和更高的性能表现。特别适合构建高并发的网络应用，如Web服务器、API网关、聊天服务器等，能够同时处理数千甚至数万个并发连接。优化资源利用异步编程减少了线程切换的开销，每个协程的内存占用通常只有几KB，而线程需要几MB。这使得内存使用效率大幅提升，可以创建数以万计的协程而不会耗尽系统资源。异步编程适用场景网络请求HTTP/HTTPS请求、API调用文件操作大文件读写、日志记录数据库操作查询、更新、批量处理实时通信WebSocket、聊天应用Python异步编程演进Python3.4：引入asyncio模块，提供事件循环、协程、任务等基础设施Python3.5：引入async/await语法，简化协程的定义和调用Python3.7+：asyncioAPI稳定，性能持续优化，生态日益完善6.2Chapter6.2协程基础掌握协程的概念与实现机制CoroutineConcept什么是协程协程定义协程（Coroutine）是一种用户态的轻量级线程，可以在单个线程内实现并发执行。与传统的线程不同，协程的调度由程序员控制，而不是由操作系统内核调度。协程可以在执行过程中挂起（suspend），保存当前状态，然后在适当的时候恢复（resume）执行，从挂起的位置继续往下执行。协程的核心特点1.协作式多任务协程主动让出CPU，而非被强制切换，避免了线程上下文切换的开销2.状态保存协程挂起时会保存执行状态（局部变量、指令指针等），恢复时继续执行3.轻量级创建和切换开销极小，一个线程内可创建数万个协程协程vs线程对比项协程线程调度方式用户态调度内核态调度切换开销极低（几十字节）较高（几MB）创建数量数万级别数百级别同步机制无需锁机制需要锁、信号量等编程模型协作式抢占式协程的优势✓高性能：切换开销小，执行效率高✓高并发：支持大量并发任务✓简化编程：避免复杂的线程同步问题✓可预测：执行顺序可控，易于调试async/awaitSyntaxPython中的async/await语法async关键字async用于定义协程函数，将普通函数转换为协程函数。调用协程函数不会立即执行，而是返回一个协程对象。asyncdeffetch_data():print("开始获取数据")awaitasyncio.sleep(1)print("数据获取完成")return"数据"#调用协程函数coroutine=fetch_data()print(type(coroutine))#await关键字await用于等待协程或可等待对象执行完成。它会暂停当前协程的执行，将控制权交还给事件循环，直到等待的对象完成。result=awaitfetch_data()print(result)#输出:数据获取完成#await也可以用于其他可等待对象awaitasyncio.sleep(2)awaitsome_task可等待对象可以被await关键字等待的对象，包括：协程对象调用asyncdef函数返回的对象Task对象使用asyncio.create_task()创建Future对象表示异步操作的最终结果重要规则1.await只能在async函数内部使用2.协程需要通过事件循环运行3.不要忘记await协程对象运行协程#方法1:使用asyncio.run()importasyncioasyncio.run(fetch_data())#方法2:获取事件循环loop=asyncio.get_event_loop()loop.run_until_complete(fetch_data())ExecutionMechanism协程的执行流程协程执行机制1协程对象创建调用asyncdef函数时，不会立即执行，而是返回一个协程对象（coroutineobject）2事件循环调度将协程对象交给事件循环（EventLoop），事件循环负责调度协程的执行3协程挂起当遇到await时，协程会挂起，将控制权交还给事件循环4协程恢复等待的操作完成后，事件循环恢复协程执行，从挂起位置继续代码示例：协程执行流程importasyncioasyncdeftask1():print("Task1:开始")awaitasyncio.sleep(1)#挂起1秒print("Task1:完成")asyncdeftask2():print("Task2:开始")awaitasyncio.sleep(2)#挂起2秒print("Task2:完成")...协程状态PENDING协程已创建但未开始执行，或正在等待调度RUNNING协程正在执行中，占用CPU资源SUSPENDED协程遇到await，暂时挂起，等待操作完成DONE协程执行完成或抛出异常，生命周期结束关键理解•单线程并发：多个协程在单线程内交替执行•非抢占式：协程主动让出CPU，不会被强制中断•事件循环调度：由事件循环统一管理和调度协程输出结果Task1:开始Task2:开始[等待1秒]Task1:完成[再等待1秒]Task2:完成总耗时:2秒6.3Chapter6.3asyncio模块详解深入理解asyncio核心组件asyncioArchitectureasyncio模块架构核心组件事件循环EventLoop是asyncio的核心，负责调度协程、处理I/O事件、管理定时任务协程Coroutine使用asyncdef定义的协程函数，是异步编程的基本单元任务Task是对协程的封装，用于调度协程的执行，是Future的子类未来对象Future表示异步操作的最终结果，可以添加回调函数组件关系图事件循环EventLoop核心调度器任务Task协程调度Future异步结果协程Coroutine异步任务编程模型与执行流程asyncio采用单线程、事件驱动的编程模型。所有协程在单个线程内并发执行，通过事件循环进行调度。执行流程①创建协程对象：调用asyncdef函数②封装为Task：使用asyncio.create_task()或asyncio.ensure_future()③注册到事件循环：Task被添加到事件循环的调度队列④事件循环调度：事件循环选择就绪的Task执行⑤协程执行：Task执行封装的协程，直到遇到await⑥协程挂起：遇到await，协程挂起，控制权交还事件循环⑦等待完成：等待await后面的操作完成⑧协程恢复：操作完成后，事件循环恢复协程执行协作关系事件循环调度中心Task协程封装Future结果占位EventLoop事件循环EventLoop事件循环的工作原理事件循环（EventLoop）是asyncio的核心机制，负责调度协程的执行、处理I/O事件、管理定时器等。它是一个无限循环，不断检查是否有就绪的任务或事件需要处理。核心职责调度协程处理I/O事件管理定时器执行回调函数事件循环的工作流程1检查就绪任务：查看就绪队列中是否有可以执行的协程2执行就绪任务：选择一个就绪的协程执行，直到遇到await3协程挂起：协程遇到await，控制权交还给事件循环4等待事件：等待I/O事件、定时器到期或新任务就绪5恢复协程：等待的操作完成，事件循环恢复协程执行事件循环基本操作#1.获取事件循环loop=asyncio.get_running_loop()#2.运行协程（推荐）asyncio.run(main())#3.创建任务task=loop.create_task(coro())#4.关闭事件循环loop.close()事件循环的优势高效调度：自动管理成千上万个协程非阻塞I/O：充分利用等待时间统一接口：简化异步编程模型事件循环调度示意图事件循环TaskATaskBTaskCTask&FutureTask对象与Future对象Task对象Task是对协程的封装，用于调度协程的执行。Task是Future的子类，它在内部封装了一个协程对象，并负责将其提交给事件循环执行。...asyncdefmy_coroutine():print("Hello")...#创建Tasktask=asyncio.create_task(my_coroutine())#或者task=asyncio.ensure_future(my_coroutine())Future对象Future表示一个异步操作的最终结果。它是一个特殊的可等待对象，代表尚未完成的操作，可以在操作完成后设置结果或异常。#创建Future对象future=asyncio.Future()#设置结果future.set_result("数据")#设置异常future.set_exception(Exception("错误"))#添加回调future.add_done_callback(callback)Task的主要方法cancel()取消任务执行，如果任务已完成则无法取消cancelled()检查任务是否被取消，返回布尔值done()检查任务是否完成，包括正常完成、取消或抛出异常result()获取任务执行结果，如果任务未完成则抛出异常TaskvsFuture关系继承关系：Task是Future的子类封装内容：Task封装协程，Future封装结果使用场景：Task用于调度，Future用于结果CommonAPIsasyncio常用APIasyncio.run()运行协程函数，创建事件循环，执行协程，完成后关闭事件循环。这是运行协程的推荐方式。asyncdefmain():print("Hello")asyncio.run(main())asyncio.create_task()创建Task对象，将协程封装为任务并调度执行。可以并发执行多个协程，不会阻塞当前协程。task1=asyncio.create_task(coro1())task2=asyncio.create_task(coro2())awaittask1...asyncio.gather()并发执行多个协程或任务，等待所有任务完成，返回结果列表。如果某个任务抛出异常，会立即传播。...awaitasyncio.gather(coro1(),coro2(),coro3())asyncio.sleep()异步版本的sleep，不会阻塞事件循环。让出CPU资源，允许其他协程执行。常用于模拟延迟或定时操作。...awaitasyncio.sleep(1)print("1秒后执行")asyncio.wait()等待多个任务完成，可以指定返回条件（ALL_COMPLETED,FIRST_COMPLETED,FIRST_EXCEPTION）。更灵活的等待控制。...done,pending=awaitasyncio.wait(tasks,return_when=asyncio.FIRST_COMPLETED)asyncio.shield()保护协程或任务不被取消。即使父协程被取消，被保护的协程仍会继续执行，直到完成。...awaitasyncio.shield(important_task)#important_task不会被取消asyncio.open_connection()创建TCP连接，返回(reader,writer)元组。用于异步网络编程，实现非阻塞的网络I/O操作。...reader,writer=awaitasyncio.open_connection('',8888)asyncio.start_server()创建TCP服务器，监听指定地址和端口。自动处理客户端连接，为每个连接创建任务。...server=awaitasyncio.start_server(handle_client,'',8888)asyncio.get_event_loop()获取当前线程的事件循环。如果当前线程没有事件循环，则创建一个新的。在异步上下文中使用。...loop=asyncio.get_event_loop()loop.run_until_complete(main())6.4Chapter6.4异步I/O操作掌握异步文件和网络I/O编程StreamsAPIStreamReader与StreamWriterStreamReader与StreamWriterStreamReader和StreamWriter是asyncio提供的高级异步I/O流对象，用于异步网络通信。它们提供了类似文件对象的接口，但完全异步非阻塞。StreamReader用于读取数据的异步流对象•read(n)-读取n个字节•readline()-读取一行•readexactly(n)-精确读取n字节StreamWriter用于写入数据的异步流对象•write(data)-写入数据•drain()-刷新缓冲区•close()-关闭连接创建连接...asyncdeftcp_client():reader,writer=awaitasyncio.open_connection('',8888)#使用reader和writer进行通信...TCP客户端完整示例importasyncioasyncdeftcp_echo_client():reader,writer=awaitasyncio.open_connection('',8888)print('Sending...')writer.write(b'Hello!')awaitwriter.drain()data=awaitreader.read(100)print(f'Received:{data.decode()}')writer.close()awaitwriter.wait_closed()asyncio.run(tcp_echo_client())关键方法说明writer.drain()必须调用，用于刷新写缓冲区，确保数据发送出去writer.wait_closed()等待连接完全关闭，异步关闭需要awaitTCPServer异步TCP服务器asyncio.start_server()asyncio.start_server()用于创建异步TCP服务器，监听指定的地址和端口，接收客户端连接。当有客户端连接时，自动为每个连接创建一个任务来执行处理函数。...server=awaitasyncio.start_server(handle_client,'',8888)客户端处理函数客户端处理函数是一个协程函数，接收reader和writer参数，用于与客户端通信。每个客户端连接都会创建一个独立的任务来执行此函数，实现并发处理。asyncdefhandle_client(reader,writer):#处理客户端连接data=awaitreader.read(100)writer.write(data)awaitwriter.drain()writer.close()完整服务器示例importasyncioasyncdefhandle_client(reader,writer):addr=writer.get_extra_info('peername')print(f"Clientconnected:{addr}")try:whileTrue:data=awaitreader.read(100)ifnotdata:breakmessage=data.decode()print(f"Received:{message}from{addr}")writer.write(data)awaitwriter.drain()exceptasyncio.CancelledError:print(f"Clientdisconnected:{addr}")finally:writer.close()awaitwriter.wait_closed()asyncdefmain():server=awaitasyncio.start_server(handle_client,'',8888)addr=server.sockets[0].getsockname()print(f'Servingon{addr}')asyncwithserver:awaitserver.serve_forever()asyncio.run(main())并发处理原理每个客户端独立任务：为每个连接创建Task单线程高并发：数千客户端同时连接非阻塞I/O：读写操作不阻塞服务器Low-LevelAPIProtocol与TransportProtocol与TransportProtocol和Transport是asyncio的低层API，提供对网络通信更精细的控制。Protocol负责协议逻辑，Transport负责数据传输。Protocolconnection_made()-连接建立data_received()-数据接收connection_lost()-连接关闭Transportwrite(data)-发送数据close()-关闭连接get_extra_info()-获取信息Protocol示例classEchoProtocol(asyncio.Protocol):defconnection_made(self,transport):self.transport=transportprint("Connectionestablished")defdata_received(self,data):print(f"Received:{data.decode()}")self.transport.write(data)defconnection_lost(self,exc):print("Connectionclosed")创建服务器...asyncdefmain():loop=asyncio.get_running_loop()server=awaitloop.create_server(lambda:EchoProtocol(),'',8888)asyncwithserver:awaitserver.serve_forever()Transport方法write(data):发送数据close():关闭连接get_extra_info():获取对等地址等信息对比：StreamsvsProtocol/Transport特性StreamsProtocol抽象层级高层API低层API编程模型线性、直观事件驱动使用难度简单较复杂6.5Chapter6.5Web应用开发使用aiohttp构建异步Web应用aiohttpFrameworkaiohttp库介绍什么是aiohttpaiohttp是基于asyncio的异步HTTP客户端/服务器框架，提供完整的Web开发功能，支持HTTP/HTTPS、WebSocket、会话管理、路由等。主要特性高性能：基于asyncio，单线程支持高并发WebSocket支持：内置WebSocket支持，实时双向通信会话管理：支持Cookie和会话，管理用户状态灵活路由：支持URL路由和视图函数安装aiohttppipinstallaiohttpaiohttpvsFlask/Django特性aiohttpFlask/Django编程模型异步同步WSGI并发能力高（单线程）依赖多线程性能优秀良好学习曲线中等简单适用场景高并发API、实时应用传统Web应用选择建议✓选择aiohttp：构建高并发API、实时应用、微服务、需要处理大量并发连接的场景✓选择Flask/Django：传统Web应用、快速原型开发、团队对同步编程更熟悉的情况HTTPClientaiohttp客户端编程ClientSessionClientSession是aiohttp客户端的核心类，用于管理HTTP连接。它支持连接复用、Cookie持久化、自定义请求头等高级功能。推荐使用asyncwith管理Session生命周期。importaiohttpimportasyncioasyncdeffetch_url(session,url):asyncwithsession.get(url)asresponse:returnawaitresponse.text()asyncdefmain():asyncwithaiohttp.ClientSession()assession:html=awaitfetch_url(session,'')print(html)常用请求方法awaitsession.get(url,params={...})awaitsession.post(url,data={...})awaitsession.put(url,json={...})高级功能示例...asyncdefadvanced_request():timeout=aiohttp.ClientTimeout(total=10)headers={'User-Agent':'MyBot/1.0'}asyncwithaiohttp.ClientSession(timeout=timeout,headers=headers)assession:asyncwithsession.get('')asresp:print(f'Status:{resp.status}')print(f'Content-Type:{resp.headers["content-type"]}')json_data=awaitresp.json()returnjson_data并发请求示例...asyncdeffetch_many(urls):asyncwithaiohttp.ClientSession()assession:tasks=[fetch_url(session,url)forurlinurls]results=awaitasyncio.gather(*tasks)returnresultsResponse对象常用方法•text()-文本•json()-JSON对象•read()-原始字节•status-状态码HTTPServeraiohttp服务器端编程Web.Applicationweb.Application是aiohttp服务器端的核心类，用于创建Web应用实例。它负责路由管理、中间件注册、应用配置等功能。fromaiohttpimportwebasyncdefhandle_request(request):returnweb.Response(text="Hello,World")app=web.Application()app.router.add_get('/',handle_request)路由定义app.router.add_get('/',handler)#GETapp.router.add_post('/',handler)#POSTapp.router.add_put('/',handler)#PUTapp.router.add_delete('/',handler)#DELETE完整Web服务器示例fromaiohttpimportwebimportasyncioasyncdefhandle_hello(request):returnweb.Response(text="Hello,World!")asyncdefhandle_user(request):user_id=request.match_info['user_id']returnweb.Response(text=f"UserID:{user_id}")asyncdefhandle_post(request):data=awaitrequest.post()returnweb.json_response({"status":"success"})definit_app():app=web.Application()app.router.add_get('/',handle_hello)app.router.add_get('/user/{user_id}',handle_user)app.router.add_post('/data',handle_post)returnappapp=init_app()web.run_app(app,host='',port=8080)请求对象Requestrequest.match_info:URL参数request.query:查询参数request.post():表单数据（需await）request.json():JSON数据（需await）WebSocketWebSocket编程WebSocket简介WebSocket是一种在单个TCP连接上进行全双工通信的协议。与HTTP的请求-响应模式不同，WebSocket允许服务端主动向客户端推送数据，实现实时双向通信。WebSocket优势低延迟：建立连接后无需重复握手双向通信：服务端可主动推送数据轻量级：数据帧头部开销小服务端WebSocket...asyncdefwebsocket_handler(request):ws=web.WebSocketResponse()awaitws.prepare(request)asyncformsginws:ifmsg.type==aiohttp.WSMsgType.TEXT:awaitws.send_str(f"Echo:{msg.data}")elifmsg.type==aiohttp.WSMsgType.ERROR:print('wsconnection