如何让Python websockets模块处理HTTP请求?是否需自定义HTTP服务器?
让Python websockets模块处理HTTP请求的实现方案
websockets模块本身是专为WebSocket协议设计的,但无需完全编写自定义HTTP服务器,有两种实用方案可以让它同时处理HTTP请求:
方案1:利用websockets的process_request参数(轻量场景)
websockets的serve方法提供了process_request参数,允许你拦截并处理HTTP请求。该参数接收一个异步函数,函数接收请求路径和请求头,返回None时会继续处理WebSocket握手;返回HTTP响应(状态码、响应头、响应体)时,会直接将响应返回给客户端。
示例代码:
import asyncio from websockets import serve from http import HTTPStatus async def websocket_handler(websocket): # 处理WebSocket消息逻辑 async for message in websocket: await websocket.send(f"Received: {message}") async def handle_http(path, request_headers): # 处理根路径的GET请求 if path == "/" and request_headers.get("Method") == "GET": response_headers = [("Content-Type", "text/plain; charset=utf-8")] return HTTPStatus.OK, response_headers, b"Hello from HTTP endpoint!" # 其他HTTP请求返回404 response_headers = [("Content-Type", "text/plain; charset=utf-8")] return HTTPStatus.NOT_FOUND, response_headers, b"404 Not Found" async def main(): async with serve( websocket_handler, "localhost", 8765, process_request=handle_http ): await asyncio.Future() # 保持服务运行 asyncio.run(main())
这种方式的优势是无需额外依赖,同一个端口同时处理WebSocket和HTTP请求,适合简单场景。
方案2:结合异步HTTP框架(复杂场景)
如果需要处理更复杂的HTTP逻辑(比如路由管理、表单解析、静态文件服务等),推荐结合aiohttp这类异步HTTP框架,将WebSocket作为框架的一个路由端点。
示例代码:
from aiohttp import web async def http_handler(request): return web.Response(text="Hello from HTTP server!") async def websocket_handler(request): ws = web.WebSocketResponse() await ws.prepare(request) async for msg in ws: if msg.type == web.WSMsgType.TEXT: await ws.send_str(f"Received: {msg.data}") elif msg.type == web.WSMsgType.ERROR: print(f"WebSocket error occurred: {ws.exception()}") return ws app = web.Application() app.add_routes([ web.get('/', http_handler), web.get('/ws', websocket_handler) ]) if __name__ == "__main__": web.run_app(app, host="localhost", port=8765)
这种方式扩展性更强,能充分利用HTTP框架的成熟生态,适合复杂应用场景。
内容的提问来源于stack exchange,提问作者BoBkiNN_
相关产品推荐
相关产品推荐

