### Python 中 SSE (Server-Sent Events) 使用方法 #### 创建 SSE 服务器 为了创建一个能够发送 Server-Sent Events 的服务器，在 Python 中可以利用 `Starlette` 或其他支持 ASGI 接口的框架。下面是一个基于 Starlette 搭建简单 SSE 服务器的例子： ```python from starlette.applications import Starlette from starlette.routing import Route from starlette.responses import StreamingResponse import asyncio async def sse_endpoint(request): async def event_publisher(): for i in range(5): # 发送五条消息给客户端 yield f"data: Message {i}\n\n" await asyncio.sleep(1) # 模拟延迟 yield "data: done\n\n" return StreamingResponse(event_publisher(), media_type="text/event-stream") routes = [ Route('/events', endpoint=sse_endpoint), ] app = Starlette(debug=True, routes=routes) ``` 此代码片段定义了一个名为 `/events` 的路由，它会每隔一秒向连接到该路径下的任何客户端广播一条新消息。 #### 构建 SSE 客户端 对于构建 SSE 客户端来说，有多种方式可以选择。这里展示如何借助于 `aiohttp` 来建立异步 HTTP 请求并监听来自上述 SSE 服务器的消息流： ```python import aiohttp import asyncio async def main(): url = 'http://localhost:8000/events' async with aiohttp.ClientSession() as session: async with session.get(url) as resp: async for line in resp.content: message = line.decode('utf-8').strip() if not message or message == ": ping": continue print(f"Received message: {message}") if __name__ == '__main__': loop = asyncio.get_event_loop() try: loop.run_until_complete(main()) finally: loop.close() ``` 这段脚本展示了怎样打开与 SSE 终结点之间的持久连接，并实时打印所接收到的数据[^4]。 #### SSE 特性和适用场景 相比于 WebSocket 技术而言，SSE 更加简洁易用，仅需标准 HTTP/HTTPS 支持即可工作，无需额外协议握手过程。然而需要注意的是，由于 SSE 是单工模式下运行的技术——即只允许由服务器主动推送给浏览器方向的信息传递，因此不适合那些需要频繁双向交互的应用场合[^3]。