基于 Sanic WebSocket 与 Valtio Proxy 构建 SvelteKit 双向状态同步层

全栈开发

文章字数: 4.2k

阅读时长: 19 分

在构建一个需要实时协作的内部数据看板时，我们面临的第一个技术挑战就是状态同步。传统的轮询或基于 REST 的刷新机制不仅效率低下，而且无法提供用户期望的即时反馈。我们需要一个稳健的、双向的、高性能的状态同步层。技术栈初选定为前端 SvelteKit，后端采用 Python 的 Sanic 框架，因其出色的异步性能。但真正的难题在于：如何设计一个与框架解耦、可独立测试、且能在前后端之间平滑同步的状态模型？

我们的初步构想是放弃将状态逻辑与 Svelte 的 stores 强绑定的做法。在真实项目中，UI 框架时常更迭，但核心业务逻辑与状态模型应当保持稳定。这就要求状态管理本身是纯粹的、框架无关的。Valtio，一个基于 Proxy 的极简状态库，进入了我们的视野。尽管它通常与 React 搭配使用，但其核心是一个不依赖任何UI库的普通 JavaScript 对象。这正是我们需要的。

最终的技术方案是：

后端 (Sanic): 维护一个权威的全局状态，并通过 WebSocket 服务管理所有客户端连接。
状态模型 (Valtio): 在一个框架无关的 JS 模块中定义应用的核心状态结构。
前端 (SvelteKit):
- 建立一个持久的 WebSocket 连接到 Sanic 服务。
- 在本地持有一个 Valtio state 的副本。
- 创建一个自定义的 Svelte store 适配器，将 Valtio 的响应式能力桥接到 Svelte 的生态中。
- 监听本地 Valtio state 的变化，生成 diff patches，并通过 WebSocket 发送给后端。
- 监听来自后端的 state 更新或 patch，并应用到本地 Valtio state。

这个方案的核心在于利用 Valtio 的 Proxy 特性实现精细化的变更追踪，并结合 WebSocket 实现高效的双向通信。

第一步：构建 Sanic WebSocket 后端

后端是整个系统的权威来源。它需要处理连接、维护全局状态，并向所有连接的客户端广播状态变更。

# file: server.py

import asyncio
import json
import logging
from uuid import uuid4

from sanic import Sanic, Request
from sanic.response import text
from sanic.websocket import WebSocketProtocol

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

app = Sanic("realtime_state_sync_app")

# 在生产环境中，这个状态应该由 Redis、数据库或其他持久化存储来管理。
# 为了演示，我们使用一个简单的内存字典作为权威状态源。
# 这个结构特意设计得复杂一些，以模拟真实应用场景。
APP_STATE = {
    "dashboardConfig": {
        "title": "Default Dashboard",
        "version": 1,
        "widgets": [
            {"id": "w1", "type": "line_chart", "position": {"x": 0, "y": 0}},
            {"id": "w2", "type": "data_table", "position": {"x": 6, "y": 0}},
        ]
    },
    "userActivity": {
        "onlineUsers": 0,
        "lastEditBy": None
    }
}

class ConnectionManager:
    """
    一个简单的连接管理器，用于跟踪所有活动的WebSocket连接。
    在真实项目中，这可能需要更复杂的实现，例如使用Redis Pub/Sub来支持多实例部署。
    """
    def __init__(self):
        self.active_connections: set = set()

    async def connect(self, websocket):
        self.active_connections.add(websocket)
        logging.info(f"New connection: {websocket.id}. Total connections: {len(self.active_connections)}")

    def disconnect(self, websocket):
        self.active_connections.remove(websocket)
        logging.info(f"Connection closed: {websocket.id}. Total connections: {len(self.active_connections)}")

    async def broadcast(self, message: str, sender=None):
        # 使用 asyncio.gather 来并发地发送消息
        tasks = [
            conn.send(message)
            for conn in self.active_connections
            if conn != sender # 避免将消息发回给发送者
        ]
        if tasks:
            await asyncio.gather(*tasks, return_exceptions=True)

manager = ConnectionManager()

@app.websocket("/ws/state")
async def state_feed(request: Request, ws: WebSocketProtocol):
    # 为每个连接分配一个唯一ID，便于日志追踪
    ws.id = uuid4()
    await manager.connect(ws)
    
    # 更新在线用户计数并广播
    APP_STATE["userActivity"]["onlineUsers"] = len(manager.active_connections)
    user_activity_update_msg = json.dumps({
        "type": "PATCH",
        "payload": {
            "path": ["userActivity"],
            "patch": {"onlineUsers": APP_STATE["userActivity"]["onlineUsers"]}
        }
    })
    await manager.broadcast(user_activity_update_msg)

    try:
        while True:
            # 监听来自客户端的消息
            raw_message = await ws.recv()
            try:
                message = json.loads(raw_message)
                msg_type = message.get("type")
                payload = message.get("payload")

                if msg_type == "REQUEST_FULL_STATE":
                    # 客户端初次连接时请求全量状态
                    logging.info(f"[{ws.id}] Received REQUEST_FULL_STATE")
                    response = {
                        "type": "FULL_STATE_UPDATE",
                        "payload": APP_STATE
                    }
                    await ws.send(json.dumps(response))

                elif msg_type == "PATCH":
                    # 客户端发送了一个状态变更补丁
                    # 这里的坑在于：必须对 payload 进行严格的验证和清理，
                    # 防止恶意客户端注入非法数据结构。在生产代码中，这里需要一个 schema 验证器。
                    if not isinstance(payload, dict) or "path" not in payload or "patch" not in payload:
                        raise ValueError("Invalid PATCH payload")

                    logging.info(f"[{ws.id}] Received PATCH: {payload}")
                    
                    # 应用补丁到权威状态
                    # 这是一个简化的实现，真实场景需要更健壮的 deep_merge 或 patch 逻辑
                    target = APP_STATE
                    path = payload["path"]
                    for key in path:
                        target = target[key]
                    
                    target.update(payload["patch"])
                    
                    # 更新元数据
                    APP_STATE["dashboardConfig"]["version"] += 1
                    APP_STATE["userActivity"]["lastEditBy"] = str(ws.id)

                    # 将确认后的补丁和元数据更新广播给所有其他客户端
                    broadcast_message = json.dumps({
                        "type": "PATCH",
                        "payload": {
                            "path": payload["path"],
                            "patch": payload["patch"]
                        }
                    })
                    meta_update_message = json.dumps({
                        "type": "PATCH",
                        "payload": {
                            "path": [], # 空路径表示根对象
                            "patch": {
                                "dashboardConfig": {"version": APP_STATE["dashboardConfig"]["version"]},
                                "userActivity": {"lastEditBy": APP_STATE["userActivity"]["lastEditBy"]}
                            }
                        }
                    })
                    
                    await manager.broadcast(broadcast_message, sender=ws)
                    await manager.broadcast(meta_update_message) # 元数据更新对所有人都可见

            except json.JSONDecodeError:
                logging.error(f"[{ws.id}] Received invalid JSON: {raw_message}")
            except (ValueError, KeyError) as e:
                logging.error(f"[{ws.id}] Error processing message: {raw_message}, error: {e}")

    except asyncio.CancelledError:
        # 连接正常关闭
        logging.info(f"[{ws.id}] Connection cancelled.")
    finally:
        manager.disconnect(ws)
        # 更新在线用户计数并广播
        APP_STATE["userActivity"]["onlineUsers"] = len(manager.active_connections)
        user_activity_update_msg = json.dumps({
            "type": "PATCH",
            "payload": {
                "path": ["userActivity"],
                "patch": {"onlineUsers": APP_STATE["userActivity"]["onlineUsers"]}
            }
        })
        await manager.broadcast(user_activity_update_msg)


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000, protocol=WebSocketProtocol, debug=True)

这个后端实现有几个关键点：

ConnectionManager: 集中管理所有 WebSocket 连接，这是实现广播的基础。
权威状态 (APP_STATE): 所有状态变更最终由服务器确认和应用。客户端只发送变更“意图”（patches），而不是直接修改本地状态后同步。
消息协议: 定义了清晰的 JSON 消息格式，包括 REQUEST_FULL_STATE, FULL_STATE_UPDATE 和 PATCH，这使得前后端通信有据可循。
错误处理与日志: 对连接生命周期和消息处理都添加了日志和基本的异常捕获，这是生产级代码的必要部分。

第二步：定义框架无关的 Valtio 状态模型

现在，我们来定义前端的状态。这个文件将是纯粹的 JavaScript，不包含任何 SvelteKit 特定的代码。

// file: src/lib/state/shared-state.js

import { proxy, subscribe } from 'valtio/vanilla';
import { devtools } from 'valtio/utils';

// 定义状态的初始结构。它应该与后端的 APP_STATE 结构保持一致，但可以为空。
const initialState = {
    dashboardConfig: {
        title: "",
        version: 0,
        widgets: []
    },
    userActivity: {
        onlineUsers: 0,
        lastEditBy: null
    },
    // 前端特有的UI状态，这部分不需要与后端同步
    ui: {
        isLoading: true,
        error: null,
        lastSyncTimestamp: null,
    }
};

// 使用 proxy 创建响应式状态对象
export const appState = proxy(initialState);

// 如果在开发环境中，连接到 Redux DevTools 以便调试
const unsubscribeDevtools = devtools(appState, { name: 'AppState' });

/**
 * 将来自服务器的全量状态合并到本地代理对象。
 * 不能直接替换 appState = new_state，这会破坏代理的引用。
 * @param {object} serverState - 从服务器获取的全量状态
 */
export function syncFullState(serverState) {
    // 这里不能简单地 Object.assign(appState, serverState)，因为它不会处理深层对象的删除和替换
    // 一个健壮的实现是逐个 key 进行赋值
    Object.keys(initialState).forEach(key => {
        if (serverState[key] !== undefined && key !== 'ui') {
            appState[key] = serverState[key];
        }
    });
    appState.ui.isLoading = false;
    appState.ui.error = null;
    appState.ui.lastSyncTimestamp = Date.now();
}

/**
 * 应用来自服务器的补丁。
 * @param {object} patchData - 包含 path 和 patch 的对象
 */
export function applyPatch(patchData) {
    const { path, patch } = patchData;

    // 同样，这里的实现需要非常健壮，以处理嵌套路径
    let target = appState;
    try {
        for (const key of path) {
            target = target[key];
        }
        Object.assign(target, patch);
        appState.ui.lastSyncTimestamp = Date.now();
    } catch (error) {
        console.error("Failed to apply patch:", patchData, error);
        appState.ui.error = "State synchronization error occurred.";
        // 在真实项目中，这里可能需要触发一次全量状态请求来修复状态不一致
    }
}

这个模块是整个前端状态管理的核心。它完全独立，可以被任何框架导入使用，甚至可以在 Node.js 环境中进行单元测试，验证 syncFullState 和 applyPatch 的逻辑是否正确。

第三步：创建 SvelteKit 与 Valtio 的桥梁

Svelte 的响应式系统基于 writable store 和 $ 语法。为了让 Svelte 组件能够“理解”Valtio proxy 的变化，我们需要创建一个适配器。

// file: src/lib/stores/valtio-adapter.js

import { readable } from 'svelte/store';
import { subscribe } from 'valtio/vanilla';
import { snapshot } from 'valtio/vanilla/utils';

/**
 * 创建一个 Svelte readable store 来订阅 Valtio proxy 的变化。
 * 当 Valtio proxy 发生变化时，这个 store 会发出一个新的快照 (snapshot)。
 * @template T
 * @param {T} proxyState - Valtio proxy 对象
 * @returns {import('svelte/store').Readable<T>} - 一个 Svelte 可读 store
 */
export function useValtio(proxyState) {
    // 使用 snapshot 获取 proxy 的当前不可变快照
    const initialValue = snapshot(proxyState);
    
    // 创建一个 readable store
    const store = readable(initialValue, (set) => {
        // 使用 Valtio 的 subscribe 函数监听 proxy 的所有变化
        const unsubscribe = subscribe(proxyState, () => {
            // 当 proxy 变化时，获取新的快照并更新 Svelte store
            set(snapshot(proxyState));
        });

        // 当 store 没有订阅者时，取消对 Valtio proxy 的监听
        return () => unsubscribe();
    });

    return store;
}

这个 useValtio 函数是关键的粘合剂。它接收一个 Valtio proxy，返回一个 Svelte store。内部通过 subscribe 监听 Valtio 的变化，并在变化时用 snapshot 创建一个全新的不可变对象来更新 Svelte store。这完全符合 Svelte 的响应式模型。

第四步：实现 WebSocket 服务与状态同步逻辑

现在我们需要一个服务来管理 WebSocket 连接，并协调状态的收发。

// file: src/lib/services/websocket-service.js

import { appState, syncFullState, applyPatch } from '$lib/state/shared-state';
import { subscribe } from 'valtio/vanilla';
import { diff } from 'valtio-json-patch';

const WEBSOCKET_URL = 'ws://localhost:8000/ws/state';

let socket = null;
let reconnectInterval = 5000; // 5秒重连间隔
let unsubscribeFromState = null;

function connect() {
    socket = new WebSocket(WEBSOCKET_URL);

    socket.onopen = () => {
        console.log('WebSocket connection established.');
        appState.ui.error = null;
        
        // 连接成功后，立即请求全量状态
        sendMessage({ type: 'REQUEST_FULL_STATE' });

        // 监听本地状态变更，并发送 patch 到服务器
        // 这是实现双向绑定的关键
        // 我们只在连接成功后才开始监听，避免在离线时产生无效的变更事件
        if (unsubscribeFromState) {
            unsubscribeFromState(); // 清理旧的订阅
        }
        
        let previousState = JSON.parse(JSON.stringify(appState)); // 深拷贝作为基线
        
        unsubscribeFromState = subscribe(appState, () => {
            const currentState = JSON.parse(JSON.stringify(appState));
            const patches = diff(previousState, currentState);
            previousState = currentState;

            // 这里的坑：UI状态的变化不应该被发送到后端
            // 我们需要过滤掉对 `ui` 字段的变更
            const filteredPatches = patches.filter(p => !p.path.includes('/ui'));

            if (filteredPatches.length > 0) {
                 // 在真实项目中，多个连续的 patch 可以被合并成一个请求
                 // 这里为了简单，我们逐个发送 (但服务器的广播逻辑已经做了优化)
                 // A better approach is to send patches in batches
                filteredPatches.forEach(patch => {
                    // valtio-json-patch 生成的 path 是 /a/b/c 格式，需要转换
                    const pathArray = patch.path.split('/').filter(Boolean);
                    
                    // 我们只处理 'replace' 和 'add' 操作，并将其统一为我们自己的 PATCH 格式
                    if (patch.op === 'replace' || patch.op === 'add') {
                         // A simple but not robust way to create patch payload
                        let patchObject = {};
                        let current = patchObject;
                        for (let i = 0; i < pathArray.length - 1; i++) {
                            current[pathArray[i]] = {};
                            current = current[pathArray[i]];
                        }
                        current[pathArray[pathArray.length - 1]] = patch.value;
                        
                        sendMessage({
                            type: 'PATCH',
                            payload: {
                                // A very simplified path logic here. Production code needs more robust path resolution.
                                path: [pathArray[0]], 
                                patch: patchObject[pathArray[0]]
                            }
                        });
                    }
                });
            }
        });
    };

    socket.onmessage = (event) => {
        try {
            const message = JSON.parse(event.data);
            switch (message.type) {
                case 'FULL_STATE_UPDATE':
                    syncFullState(message.payload);
                    break;
                case 'PATCH':
                    applyPatch(message.payload);
                    break;
                default:
                    console.warn('Received unknown message type:', message.type);
            }
        } catch (error) {
            console.error('Error processing message from server:', error);
        }
    };

    socket.onclose = (event) => {
        console.error('WebSocket connection closed. Attempting to reconnect...', event.reason);
        if (unsubscribeFromState) {
            unsubscribeFromState();
            unsubscribeFromState = null;
        }
        setTimeout(connect, reconnectInterval);
    };

    socket.onerror = (error) => {
        console.error('WebSocket error:', error);
        appState.ui.error = 'Connection to server failed.';
        socket.close(); // 这会触发 onclose 中的重连逻辑
    };
}

function sendMessage(message) {
    if (socket && socket.readyState === WebSocket.OPEN) {
        socket.send(JSON.stringify(message));
    } else {
        console.warn('WebSocket is not open. Message not sent:', message);
    }
}

// 自动初始化连接
if (typeof window !== 'undefined') {
    connect();
}

这段代码处理了 WebSocket 的完整生命周期，包括自动重连。最精妙的部分在于 onopen 回调中对本地 appState 的订阅。我们使用 valtio-json-patch 的 diff 工具来计算状态变化前后的差异，然后将这些差异（patches）发送给服务器。这远比每次都发送整个状态对象要高效得多。

第五步：在 Svelte 组件中使用

所有基础设施都已就位，现在在 Svelte 组件中使用它就变得非常简单了。

<!-- file: src/routes/+page.svelte -->

<script>
    import { appState } from '$lib/state/shared-state';
    import { useValtio } from '$lib/stores/valtio-adapter';

    // 使用我们的适配器将 Valtio proxy 转换为 Svelte store
    const state = useValtio(appState);

    function handleTitleChange(event) {
        // 直接修改 Valtio proxy 对象
        // 我们的 websocket-service 会自动侦测到这个变化并发送 patch
        appState.dashboardConfig.title = event.target.value;
    }

    function addWidget() {
        // SvelteKit/Vite HMR 可能会导致这里的状态变更被发送多次，
        // 在真实开发中需要注意。
        const newWidget = {
            id: `w${Date.now()}`,
            type: 'new_chart',
            position: { x: 0, y: Math.max(...$state.dashboardConfig.widgets.map(w => w.position.y), -1) + 2 }
        };
        // 直接修改 Valtio proxy
        appState.dashboardConfig.widgets.push(newWidget);
    }
</script>

<main class="container">
    {#if $state.ui.isLoading}
        <p>Connecting to server and fetching state...</p>
    {:else if $state.ui.error}
        <p class="error">Error: {$state.ui.error}</p>
    {:else}
        <header>
            <h1>Dashboard Configuration</h1>
            <div class="meta-info">
                <span>Version: {$state.dashboardConfig.version}</span>
                <span>Online Users: {$state.userActivity.onlineUsers}</span>
                <span>Last Edit By: {$state.userActivity.lastEditBy || 'N/A'}</span>
            </div>
        </header>

        <div class="form-group">
            <label for="dashboardTitle">Dashboard Title</label>
            <input
                id="dashboardTitle"
                type="text"
                value={$state.dashboardConfig.title}
                on:input={handleTitleChange}
            />
        </div>

        <div class="widgets-section">
            <h2>Widgets</h2>
            <button on:click={addWidget}>Add Widget</button>
            <ul>
                {#each $state.dashboardConfig.widgets as widget (widget.id)}
                    <li>
                        ID: {widget.id}, Type: {widget.type}, Position: (x: {widget.position.x}, y: {widget.position.y})
                    </li>
                {/each}
            </ul>
        </div>
    {/if}
</main>

<style>
    .container { max-width: 800px; margin: 2rem auto; font-family: sans-serif; }
    .meta-info { display: flex; gap: 1rem; color: #555; font-size: 0.9em; margin-bottom: 1.5rem; }
    .form-group { margin-bottom: 1rem; }
    .widgets-section { margin-top: 2rem; }
    .error { color: red; }
</style>

组件代码异常简洁。它不知道 WebSocket 的存在，也不知道状态同步的复杂细节。它只是通过 $state 响应式地展示数据，并通过直接修改 appState 对象来更新数据。所有的同步逻辑都被封装在了底层的服务和状态模块中。

以下是整个数据流的概览：

sequenceDiagram
    participant User
    participant SvelteComponent
    participant ValtioProxy
    participant WebSocketService
    participant SanicServer
    participant OtherClients

    User->>SvelteComponent: 修改输入框 (e.g., Dashboard Title)
    SvelteComponent->>ValtioProxy: appState.dashboardConfig.title = "New Title"
    ValtioProxy-->>WebSocketService: subscribe() 触发回调
    WebSocketService->>WebSocketService: diff(oldState, newState) 生成 patch
    WebSocketService->>SanicServer: send(type: "PATCH", payload: {...})
    SanicServer->>SanicServer: 应用 patch 到权威状态
    SanicServer->>OtherClients: broadcast(type: "PATCH", payload: {...})
    
    OtherClients->>OtherClients: onmessage() 接收到 patch
    OtherClients->>OtherClients: applyPatch() 更新其本地 ValtioProxy
    OtherClients-->>OtherClients: Valtio 变化触发 Svelte store 更新, UI 自动刷新

局限性与未来迭代路径

这套方案虽然优雅地解决了前后端状态同步和框架解耦的问题，但在生产环境中仍有需要完善之处。

首先，当前的 PATCH 逻辑是“最后写入者获胜”（Last Write Wins），没有处理并发修改的冲突。如果两个用户同时修改同一个字段，后到达服务器的 patch 会覆盖前一个。对于需要更强一致性的协作场景，可以引入 CRDTs (Conflict-free Replicated Data Types) 或 OT (Operational Transformation) 算法，但这会极大地增加系统的复杂性。

其次，后端的状态是单实例内存存储，无法水平扩展。在生产环境中，需要将 APP_STATE 和 ConnectionManager 的逻辑迁移到 Redis 或类似的分布式缓存/消息队列中，利用其 Pub/Sub 功能来实现跨多个 Sanic 实例的广播。

最后，补丁的生成和应用逻辑可以进一步优化。例如，valtio-json-patch 生成的 patch 格式与 RFC 6902 标准一致，我们的自定义实现做了简化。一个更健壮的系统应该在前后端都使用标准的 JSON Patch 库来保证一致性。同时，对 WebSocket 消息进行二进制序列化（如 MessagePack 或 Protobuf）可以进一步降低网络负载。

WebSocket SvelteKit Valtio Sanic 状态管理

构建支持动态模型切换的核心库：融合DDD、MLflow与Consul服务发现

2023-10-27 架构与设计

Consul 核心库 Domain-Driven Design 服务发现 MLflow

基于MyBatis数据变更事件实现Gatsby站点的增量静态生成

2023-10-27 后端架构

MyBatis Backend Gatsby 性能优化事件驱动