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

Uvicorn源码中的中介者模式：组件解耦与通信中心

news 2026/5/12 5:41:49

Uvicorn源码中的中介者模式：组件解耦与通信中心

【免费下载链接】uvicornAn ASGI web server, for Python. 🦄项目地址: https://gitcode.com/GitHub_Trending/uv/uvicorn

Uvicorn作为Python高性能ASGI Web服务器，在其架构设计中巧妙地运用了中介者模式来实现组件间的解耦与高效通信。本文将深入分析Uvicorn源码中的中介者模式实现，揭示这一设计模式如何帮助构建高并发、可扩展的Web服务器架构。

中介者模式在Uvicorn中的核心体现

Uvicorn的中介者模式主要体现在ServerState类和Config类中，它们作为中央协调者，管理着服务器中各个组件之间的通信与状态共享。

ServerState：全局状态的中介者

在uvicorn/server.py中，ServerState类扮演着关键的中介者角色：

class ServerState: """ Shared servers state that is available between all protocol instances. """ def __init__(self) -> None: self.total_requests = 0 self.connections: set[Protocols] = set() self.tasks: set[asyncio.Task[None]] = set() self.default_headers: list[tuple[bytes, bytes]] = []

ServerState的核心职责包括：

跟踪所有活跃连接（connections）
管理后台任务（tasks）
记录总请求数（total_requests）
维护默认HTTP头部信息

中介者模式如何实现组件解耦

1. 协议实例与状态管理的分离

在HTTP协议实现中（如uvicorn/protocols/http/h11_impl.py），每个协议实例通过中介者访问共享状态：

class H11Protocol(asyncio.Protocol): def __init__( self, config: Config, server_state: ServerState, app_state: dict[str, Any], _loop: asyncio.AbstractEventLoop | None = None, ) -> None: # ... self.server_state = server_state self.connections = server_state.connections self.tasks = server_state.tasks

这种设计使得协议实例无需直接相互通信，所有协调工作都通过ServerState中介者完成。

2. 连接管理的集中控制

中介者模式让连接管理变得集中而高效。当新连接建立时：

def connection_made(self, transport: asyncio.Transport) -> None: self.connections.add(self) # 通过中介者注册连接

当连接断开时：

def connection_lost(self, exc: Exception | None) -> None: self.connections.discard(self) # 通过中介者注销连接

3. 任务生命周期的协调

Uvicorn中的后台任务也通过中介者进行管理：

# 添加任务到中介者 self.tasks.add(task) # 任务完成后自动清理 task.add_done_callback(self.tasks.discard)

中介者模式带来的架构优势

1.降低组件间耦合度

协议实例（HTTP/WebSocket）之间无需直接引用
每个组件只需与中介者交互
新增协议实现更加容易

2.简化并发控制

连接限制检查统一处理
任务超时管理集中化
资源清理逻辑统一

3.提高可维护性

状态管理逻辑集中在一处
调试和监控更加方便
扩展新功能时影响范围可控

实际应用场景分析

场景一：连接数限制检查

在h11_impl.py中，当处理新请求时，通过中介者检查当前连接数：

if self.limit_concurrency is not None and ( len(self.connections) >= self.limit_concurrency or len(self.tasks) >= self.limit_concurrency ): # 返回503服务不可用

场景二：优雅关闭机制

服务器关闭时，通过中介者通知所有连接和任务：

async def shutdown(self, sockets: list[socket.socket] | None = None) -> None: # 请求所有现有连接关闭 for connection in list(self.server_state.connections): connection.shutdown() # 等待所有任务完成 while self.server_state.tasks and not self.force_exit: await asyncio.sleep(0.1)

场景三：默认头部信息同步

服务器定期更新默认头部信息，所有协议实例通过中介者获取最新值：

# 在Server的on_tick方法中更新 self.server_state.default_headers = date_header + self.config.encoded_headers # 协议实例中使用 default_headers=self.server_state.default_headers