Sanic 22.6 Websocket协议如何检查WebSocket连接是否打开

Sanic 22.6 对WebSocket协议层做了重构，移除了早期版本WebsocketImplProtocol实例上直接暴露的公开状态查询属性，可通过以下三种可行方案判断实时连接状态：

方法1：读取实例内部状态字段（快速判断） #

WebsocketImplProtocol实例保留了内部状态枚举字段state，可直接读取判断，状态值对应关系如下：

• 0：连接已打开（OPEN），可正常执行收发操作
• 1：连接正在关闭（CLOSING），关闭握手流程已触发
• 2：连接已关闭（CLOSED），连接资源已释放

示例代码：

@app.websocket("/ws")
async def ws_endpoint(request, ws):
    match ws.state:
        case 0:
            # 连接正常逻辑
            await ws.send("connected")
        case 1:
            # 正在关闭逻辑
            print("connection closing")
        case 2:
            # 已关闭逻辑
            print("connection closed")

方法2：收发操作异常捕获（生产环境推荐） #

直接读取内部字段属于依赖非公开实现的方案，版本迭代可能出现兼容问题，生产环境最可靠的方式是通过收发操作的返回/异常判断状态：

• 调用await ws.recv()时，若连接已正常关闭会返回None，异常断开会抛出ConnectionClosed异常
• 调用await ws.send()时，若连接不可用会直接抛出ConnectionClosed异常

示例代码：

from websockets.exceptions import ConnectionClosed

@app.websocket("/ws")
async def ws_endpoint(request, ws):
    try:
        while True:
            msg = await ws.recv()
            if msg is None:
                # 正常关闭
                break
            await ws.send(f"received: {msg}")
    except ConnectionClosed:
        # 异常断开
        print("connection closed unexpectedly")

方法3：通过底层asyncio传输对象判断 #

WebsocketImplProtocol实例暴露了底层asyncio的传输对象transport，可通过该对象的公开方法判断状态：

• 若ws.transport is None，说明连接已完全释放
• 若ws.transport.is_closing()返回True，说明连接处于关闭流程中
• 其余情况连接处于正常打开状态

封装判断函数示例：

def check_ws_alive(ws) -> bool:
    if not ws.transport:
        return False
    return not ws.transport.is_closing()

注意：方法1的state字段不属于Sanic公开API范畴，仅适合临时快速验证场景，长期维护的业务优先选择方法2的异常捕获方案，跨版本兼容性最好。

内容的提问来源于stack exchange，提问作者giwyni