如何在关闭PyWebView窗口时无错误地优雅停止Sanic服务器（避免WebSocket关闭报错）

我完全理解你的困扰——既不想用hack性质的猴子补丁，也不想每次关闭窗口都看到烦人的RuntimeError。这个错误本质上是因为PyWebView窗口关闭时，Sanic还在尝试对已经被强制断开的WebSocket连接的TCP传输执行操作。我们可以通过优化关闭流程、主动管理连接和利用Sanic内置的优雅关闭机制来解决这个问题，不需要修改Sanic源码。

核心解决思路 #

1. 确保前端主动关闭WebSocket：让浏览器在窗口关闭时先主动发送关闭帧，给Sanic足够时间处理断开
2. 主动清理活跃连接：在停止服务器前，主动关闭所有剩余的WebSocket连接，避免Sanic后续操作无效资源
3. 有序触发优雅关闭：给PyWebView绑定关闭回调，触发一个按顺序的清理流程，再停止Sanic
4. 利用Sanic内置机制：使用app.stop(graceful=True)触发内置的优雅关闭，等待所有请求和连接处理完成

修改后的完整示例代码 #

这个版本整合了所有优化点，关闭窗口时不会再出现RuntimeError：

#!/usr/bin/env python3
# main.py
# -----------------------------------------------------------------------------
# Suppress pkg_resources warning from tracerite/html
# -----------------------------------------------------------------------------
import warnings
warnings.filterwarnings(
    "ignore", message="pkg_resources is deprecated as an API.*", module="tracerite.html",
)

import argparse
import asyncio
import multiprocessing
import os
import sys
from sanic import Sanic, Websocket
from sanic.response import html
from websockets.exceptions import ConnectionClosedOK, ConnectionClosedError
import webview

# -----------------------------------------------------------------------------
# Defaults
# -----------------------------------------------------------------------------
DEFAULT_PORT = 8000
HOST = "127.0.0.1"

# -----------------------------------------------------------------------------
# App setup
# -----------------------------------------------------------------------------
app = Sanic("ShellApp")
app.config.DEBUG = True
app.config.AUTO_RELOAD = True
app.config.UI_PORT = DEFAULT_PORT
clients = set()
app.ctx.tasks = []

# -----------------------------------------------------------------------------
# HTTP & WebSocket handlers
# -----------------------------------------------------------------------------
@app.route("/")
async def index(request):
    return html(
        """<!DOCTYPE html>
        <html>
        <head><title>Sanic Shell</title></head>
        <body>
        <h1>Messages (port {port})</h1>
        <ul id="messages"></ul>
        <script>
        const ws = new WebSocket(`ws://${window.location.host}/ws`);
        ws.onmessage = e => {{
            const li = document.createElement("li");
            li.textContent = e.data;
            document.getElementById("messages").appendChild(li);
        }};
        // 增强的WebSocket主动关闭逻辑
        window.onbeforeunload = async () => {{
            if (ws.readyState === WebSocket.OPEN) {{
                ws.close(1000, "User closed window");
                // 给100ms让关闭帧发送完成
                await new Promise(resolve => setTimeout(resolve, 100));
            }}
        }};
        </script>
        </body>
        </html>""".format(port=app.config.UI_PORT)
    )

@app.websocket("/ws")
async def websocket_handler(request, ws: Websocket):
    clients.add(ws)
    try:
        async for _ in ws:
            pass
    except (ConnectionClosedOK, ConnectionClosedError, asyncio.CancelledError, RuntimeError):
        # 捕获所有可能的关闭异常，避免未处理报错
        pass
    finally:
        if ws in clients:
            clients.discard(ws)

# -----------------------------------------------------------------------------
# Graceful Shutdown Logic
# -----------------------------------------------------------------------------
async def graceful_shutdown(app: Sanic):
    # 1. 标记应用停止，让后台任务退出循环
    app.ctx.running = False

    # 2. 主动关闭所有活跃WebSocket连接
    for ws in list(clients):
        try:
            if not ws.closed:
                await ws.close(code=1000, reason="Server shutting down gracefully")
                await asyncio.sleep(0.1)
        except Exception:
            # 忽略关闭时的异常，防止触发新错误
            pass
    clients.clear()

    # 3. 清理后台任务
    for t in app.ctx.tasks:
        if not t.done():
            t.cancel()
    await asyncio.gather(*app.ctx.tasks, return_exceptions=True)
    app.ctx.tasks.clear()

    # 4. 触发Sanic内置优雅关闭
    await app.stop(graceful=True, timeout=5)

def on_webview_closed():
    # PyWebView关闭时触发优雅关闭（跨线程调用异步函数）
    asyncio.run_coroutine_threadsafe(graceful_shutdown(app), app.loop)

# -----------------------------------------------------------------------------
# Lifecycle hooks
# -----------------------------------------------------------------------------
@app.listener("before_server_start")
async def setup(app, loop):
    app.ctx.running = True
    # 启动后台任务
    t1 = loop.create_task(tick_worker(app))
    app.ctx.tasks.append(t1)

@app.listener("before_server_stop")
async def cleanup(app, loop):
    # 兼容重复调用，确保清理逻辑安全
    if app.ctx.running:
        app.ctx.running = False
    for t in app.ctx.tasks:
        if not t.done():
            t.cancel()
    await asyncio.gather(*app.ctx.tasks, return_exceptions=True)
    app.ctx.tasks.clear()
    clients.clear()

# -----------------------------------------------------------------------------
# Background tasks
# -----------------------------------------------------------------------------
async def tick_worker(app):
    count = 0
    while app.ctx.running:
        for ws in list(clients):
            try:
                await ws.send(f"tick {count}")
            except Exception:
                # 发送失败自动移除无效连接
                clients.discard(ws)
        count += 1
        await asyncio.sleep(1)

# -----------------------------------------------------------------------------
# PyWebView Startup
# -----------------------------------------------------------------------------
def start_webview(port):
    window = webview.create_window("Sanic Shell", f"http://{HOST}:{port}")
    webview.start(on_webview_closed, debug=True)

# -----------------------------------------------------------------------------
# Main Entry
# -----------------------------------------------------------------------------
if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--ui", action="store_true", help="Start PyWebView UI")
    parser.add_argument("--port", type=int, default=DEFAULT_PORT, help="Server port")
    args = parser.parse_args()

    app.config.UI_PORT = args.port

    if args.ui:
        # 用多进程分离Sanic和PyWebView，避免线程阻塞问题
        server_process = multiprocessing.Process(
            target=lambda: app.run(host=HOST, port=args.port, single_process=True)
        )
        server_process.start()
        try:
            start_webview(args.port)
        finally:
            # 确保服务器进程正常退出
            server_process.terminate()
            server_process.join()
    else:
        app.run(host=HOST, port=args.port, single_process=True)

关键修改说明 #

1. 前端WebSocket优化：在window.onbeforeunload中主动发送关闭帧，并等待100ms确保帧发送完成，避免Sanic收到强制断开的连接
2. 异常捕获增强：WebSocket handler新增捕获RuntimeError，覆盖传输已关闭的场景
3. 优雅关闭流程：新增graceful_shutdown函数，按顺序标记应用停止、关闭所有WebSocket、清理后台任务，最后触发Sanic内置优雅关闭
4. PyWebView回调绑定：窗口关闭时触发优雅关闭，而不是直接终止进程
5. 多进程分离：Sanic和PyWebView分进程运行，避免线程阻塞导致的清理不及时问题

这样调整后，关闭PyWebView窗口时，Sanic会有序完成所有清理工作，不会再出现RuntimeError，完全不需要猴子补丁。

内容来源于stack exchange