GraphQL 订阅驱动链上事件流：WebSocket 重连、去重与断线补偿策略

一、链上事件不能靠轮询维持

典型的 DApp 后端获取链上事件的做法是轮询：每 N 秒调用eth_getLogs或getProgramAccounts，把新事件写入数据库，然后通知前端。这种模式在事件量低的场景能用，但面对高频合约（如 DEX、NFT Marketplace、链上游戏），会碰到两个问题：一是轮询间隔内的事件堆积导致前端延迟，二是网络抖动时容易丢块。

更合理的做法是用 GraphQL Subscription 做事件驱动的推送链路：合约 emit 事件 → 索引层（如 The Graph / SubQuery / 自建）解析并写入 → GraphQL 服务通过 WebSocket 推送给订阅者。前端只需要维护一个 WebSocket 连接，不用再写轮询逻辑。

但这套链路在生产环境有三个坑：WebSocket 断开后怎么自动重连、网络抖动导致的消息重复怎么去重、离线期间丢失的事件怎么补偿。

二、链路架构：从合约事件到前端渲染

sequenceDiagram participant C as 智能合约 participant I as 索引层<br/>(The Graph / SubQuery) participant G as GraphQL 服务 participant W as WebSocket Gateway participant F as 前端 DApp C->>I: emit Transfer(from, to, amount) I->>G: 写入数据库 + 触发订阅 G->>W: 推送事件数据 W->>F: subscription { transferEvents } F->>F: 去重检查 + 状态更新 Note over W,F: 断开时 F->>W: 重连 (带 lastEventId) W->>G: 查询丢失事件 G->>W: 补偿推送 W->>F: 离线期间的 events[]

核心设计点：

• 去重在消费端做，不在推送端。推送端只需保证 at-least-once，消费端用事件 ID 去重。
• 断线补偿用游标（cursor）做，每次处理完一批事件就把最新的 cursor 持久化到 localStorage。重连时从游标位置续订。
• WebSocket 重连要带退避策略，不能在断线后立即无限重试打爆服务器。

三、生产级实现

服务端：GraphQL Subscription + 事件总线

用graphql-ws协议（非subscriptions-transport-ws，后者已废弃）实现服务端推送：

// server/subscription/chainEvents.ts import { PubSub } from "graphql-subscriptions"; import { withFilter } from "graphql-subscriptions"; // 使用 Redis PubSub 替代内存 PubSub——跨进程/跨实例共享 // graphql-redis-subscriptions 支持多实例水平扩展 const pubsub = new PubSub(); interface TransferEvent { eventId: string; // 全局唯一事件 ID: `${blockNumber}-${logIndex}` blockNumber: number; transactionHash: string; from: string; to: string; tokenId: string; amount: string; timestamp: number; } // 订阅类型定义 export const typeDefs = ` type TransferEvent { eventId: String! blockNumber: Int! transactionHash: String! from: String! to: String! tokenId: String! amount: String! timestamp: Int! } type Subscription { transferEvents( contractAddress: String! fromBlock: Int ): TransferEvent! } `; // Resolver export const resolvers = { Subscription: { transferEvents: { // 订阅时触发——创建新的事件监听 subscribe: withFilter( () => pubsub.asyncIterator(["TRANSFER_EVENT"]), (payload: { transferEvents: TransferEvent }, variables: { contractAddress: string; fromBlock?: number }) => { // 只推送指定合约的事件 // 注意：这里 contractAddress 是索引层的逻辑标识，不是链上地址 // 实际场景中从 payload 的 event 数据中提取 if (variables.fromBlock) { return payload.transferEvents.blockNumber >= variables.fromBlock; } return true; } ), }, }, }; /** * 事件索引入库后调用此函数推送 * * 设计决策： * - eventId 用 `${blockNumber}-${logIndex}` 保证全局唯一，去重依赖它 * - 推送是 fire-and-forget：即使某个订阅者离线，消息也不会丢失 * （索引层会持久化所有事件，断线补偿从数据库查） */ export async function publishTransferEvent(event: TransferEvent) { await pubsub.publish("TRANSFER_EVENT", { transferEvents: event, }); }

客户端：WebSocket 重连 + 去重 + 断线补偿

// lib/subscription/EventStreamClient.ts type EventHandler<T> = (event: T) => void | Promise<void>; interface StreamConfig { url: string; // GraphQL WebSocket endpoint subscription: string; // GraphQL subscription query variables?: Record<string, any>; maxRetries?: number; // 最大重连次数，默认无限 retryBaseDelay?: number; // 退避基础延迟 ms，默认 1000 retryMaxDelay?: number; // 退避最大延迟 ms，默认 30000 heartbeatInterval?: number; // 心跳间隔 ms，默认 10000 } interface StreamState { connected: boolean; lastEventId: string | null; retryCount: number; lastHeartbeat: number; } /** * 事件流客户端 * * 核心设计： * 1. 自动重连 + 指数退避（避免重连风暴） * 2. 基于 eventId 的去重（at-least-once 语义） * 3. 心跳检测（及早发现死连接，不等到 TCP 超时） * 4. 游标断线补偿（重连后拉取丢失的事件） */ export class EventStreamClient<T extends { eventId: string }> { private ws: WebSocket | null = null; private state: StreamState = { connected: false, lastEventId: null, retryCount: 0, lastHeartbeat: 0, }; private handler: EventHandler<T> | null = null; private eventIdSet: Set<string> = new Set(); private disconnectResolve: (() => void) | null = null; private config: Required<StreamConfig>; constructor(config: StreamConfig) { this.config = { maxRetries: Infinity, retryBaseDelay: 1000, retryMaxDelay: 30000, heartbeatInterval: 10000, ...config, }; } /** * 连接并开始接收事件 * * 为什么手动实现 WebSocket 而不是用 graphql-ws 的 createClient？ * 因为 graphql-ws 的自带重连不处理"重连期间丢失的事件补偿"， * 我们需要在重连后查询数据库拉取离线期间的事件。 */ connect(handler: EventHandler<T>): Promise<void> { this.handler = handler; return this.establishConnection(); } private async establishConnection(): Promise<void> { // 从 localStorage 读取上次处理到的 eventId 作为断线补偿游标 const cursor = this.getCursor(); // 重连前先补偿丢失的事件——优先保证数据完整性 if (cursor && this.state.lastEventId) { await this.fetchMissingEvents(cursor, this.state.lastEventId); } // graphql-ws 协议的连接初始化消息 // 注意：使用 graphql-ws 协议的 GQL_CONNECTION_INIT this.ws = new WebSocket(this.config.url, "graphql-transport-ws"); this.ws.onopen = () => { this.state.connected = true; this.state.retryCount = 0; this.state.lastHeartbeat = Date.now(); // 发送连接初始化（graphql-ws 协议要求） this.sendMessage({ type: "connection_init" }); // 发送订阅请求 // 使用 fromCursor 参数告诉服务端从哪个位置开始推送 this.sendMessage({ id: "1", type: "subscribe", payload: { query: this.config.subscription, variables: { ...this.config.variables, fromBlock: cursor ? this.parseCursorToBlock(cursor) : undefined, }, }, }); this.startHeartbeat(); }; this.ws.onmessage = (event) => { const message = JSON.parse(event.data); switch (message.type) { case "connection_ack": // 连接确认，可以开始接数据 break; case "next": this.handleEvent(message.payload.data); break; case "error": console.error("Subscription error:", message.payload); // 业务错误不触发重连，只有网络错误才重连 break; case "complete": // 服务端主动关闭订阅 this.cleanup(); break; } }; this.ws.onclose = (event) => { this.state.connected = false; // 非正常关闭才重连（code 1000 是正常关闭） if (event.code !== 1000) { this.scheduleReconnect(); } }; this.ws.onerror = () => { // onerror 后一定会触发 onclose，这里不处理 }; // 等待连接建立（带超时） await new Promise<void>((resolve, reject) => { const timeout = setTimeout(() => { reject(new Error("WebSocket connection timeout")); }, 10000); const check = setInterval(() => { if (this.state.connected) { clearTimeout(timeout); clearInterval(check); resolve(); } }, 100); }); } /** * 处理收到的事件——去重 + 游标更新 */ private async handleEvent(data: any) { // graphql-ws 的 next 消息中，data 直接是 payload const event = data.transferEvents || data; if (!event?.eventId) return; // 去重：基于 eventId // Set 内存会增长，生产环境建议用 LRU Cache（如 lru-cache 库）限制上限 if (this.eventIdSet.has(event.eventId)) return; this.eventIdSet.add(event.eventId); this.state.lastEventId = event.eventId; // 持久化游标——如果此时浏览器崩溃，重启后能从这里续订 this.saveCursor(event.eventId); // 调用业务处理器 if (this.handler) { try { await this.handler(event as T); } catch (err) { // 业务处理失败不阻塞流——记录错误日志 console.error("Event handler error:", err); } } // 清理过旧的事件 ID，防止内存泄漏 // 简单策略：保留最近 10000 个 if (this.eventIdSet.size > 10000) { const toDelete = Array.from(this.eventIdSet).slice(0, 5000); toDelete.forEach((id) => this.eventIdSet.delete(id)); } } /** * 断线补偿：从索引层查询丢失的事件 * * 设计决策： * - 不用 WebSocket 做补偿（WebSocket 是推送通道，不是查询通道） * - 走 HTTP GraphQL query 查询两个游标之间的事件 * - 超时设短一些（5s），补偿失败不影响主流程重连 */ private async fetchMissingEvents(fromCursor: string, toCursor: string) { try { const response = await fetch(this.config.url.replace("ws", "http"), { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ query: ` query MissingEvents($from: String!, $to: String!) { eventsBetween(from: $from, to: $to) { eventId blockNumber from to amount } } `, variables: { from: fromCursor, to: toCursor }, }), signal: AbortSignal.timeout(5000), }); const { data } = await response.json(); if (data?.eventsBetween) { for (const event of data.eventsBetween) { // 补偿事件也走 handler 处理——保证有序 if (this.handler && !this.eventIdSet.has(event.eventId)) { this.eventIdSet.add(event.eventId); await this.handler(event as T); } } } } catch (err) { // 补偿失败不阻塞重连——至少当前连接是活的 console.warn("Failed to fetch missing events, continuing with reconnect", err); } } /** * 指数退避重连 * * 退避公式：min(baseDelay × 2^retryCount, maxDelay) * 第 1 次：1s 第 5 次：16s 第 10 次：最大 30s */ private scheduleReconnect() { if (this.state.retryCount >= this.config.maxRetries) { console.error("Max retries reached, giving up"); return; } const delay = Math.min( this.config.retryBaseDelay * Math.pow(2, this.state.retryCount), this.config.retryMaxDelay ); setTimeout(() => { this.state.retryCount++; this.establishConnection().catch(() => { // establishConnection 内部会触发 scheduleReconnect }); }, delay); } /** * 心跳检测——在 WebSocket 层面发现死连接 * * 设计决策：用 ping/pong 而不是 graphql-ws 的 connection_keep_alive * 因为 ping/pong 是 WebSocket 协议层的，由浏览器自动回复 pong */ private startHeartbeat() { const interval = setInterval(() => { if (!this.ws || this.ws.readyState !== WebSocket.OPEN) { clearInterval(interval); return; } // 上次收到消息超过心跳间隔 2 倍——可能死连接 if (Date.now() - this.state.lastHeartbeat > this.config.heartbeatInterval * 2) { this.ws.close(4001, "Heartbeat timeout"); clearInterval(interval); return; } this.ws.send(JSON.stringify({ type: "ping" })); }, this.config.heartbeatInterval); } disconnect() { return new Promise<void>((resolve) => { this.disconnectResolve = resolve; if (this.ws) { this.ws.close(1000, "Client disconnect"); // 1000 = 正常关闭 } else { resolve(); } }); } private sendMessage(message: Record<string, unknown>) { if (this.ws?.readyState === WebSocket.OPEN) { this.ws.send(JSON.stringify(message)); } } private cleanup() { this.state.connected = false; // 清理资源：心跳定时器、事件 ID 集合 this.eventIdSet.clear(); } private getCursor(): string | null { try { return localStorage.getItem("event_stream_cursor"); } catch { return null; } } private saveCursor(eventId: string) { try { localStorage.setItem("event_stream_cursor", eventId); } catch { // localStorage 满了或其他异常——不影响主流程 } } private parseCursorToBlock(cursor: string): number { // eventId 格式: `${blockNumber}-${logIndex}` return parseInt(cursor.split("-")[0], 10) || 0; } }

使用示例

// hooks/useTransferEvents.ts import { useEffect, useRef } from "react"; import { EventStreamClient } from "@/lib/subscription/EventStreamClient"; export function useTransferEvents( contractAddress: string, onEvent: (event: TransferEvent) => void ) { const clientRef = useRef<EventStreamClient<TransferEvent> | null>(null); useEffect(() => { const client = new EventStreamClient<TransferEvent>({ url: process.env.NEXT_PUBLIC_GRAPHQL_WS!, subscription: ` subscription TransferEvents($contractAddress: String!) { transferEvents(contractAddress: $contractAddress) { eventId blockNumber from to tokenId amount timestamp } } `, variables: { contractAddress }, }); clientRef.current = client; client.connect(onEvent).catch(console.error); return () => { client.disconnect(); }; }, [contractAddress]); return clientRef; }

四、边界分析：推送不是万能药

适用场景：

• 事件频率中等的 DApp（每秒 10-1000 个事件）。太低频收益不如轮询，太高频 WebSocket 信道会阻塞
• 需要准实时更新的前端（订单簿、拍卖出价、链上消息）
• 多客户端共享同一事件源（GraphQL Subscription 天然支持多订阅者）

不适用场景：

• 毫秒级高频事件（如链上游戏每帧状态）。WebSocket 的单信道特性无法承载，应考虑 UDP + 状态差量同步
• 对数据完整性要求 100% 的场景。WebSocket 推送 + 补偿查询能接近但无法保证完全一致——链重组可能使blockNumber-logIndex作废

关键限制：

• 去重 Set 在长时间运行的 SPA 中会无限增长。生产环境应使用固定大小的 LRU Cache
• localStorage 的游标在跨设备场景失效。用户切换到手机时游标不存在，补偿查询会从 0 开始拉取，可能产生大量数据
• WebSocket 连接在移动网络（4G/5G）上频繁切换基站时会反复断开重建，指数退避的上限需要根据场景调低

事件顺序性问题：

GraphQL Subscription 的推送顺序在单个 WebSocket 连接中是保证的，但在断线重连后可能发生变化。游标补偿查询拉取的是批量结果，不保证"重连后收到的第一条实时推送"和"补偿查询中最后一条"之间的顺序关系。实际代码中可能出现这样的情况：补偿查询的最后一个事件（按游标顺序应该在前面）在实时推送的第一个事件之后才被处理。解决思路是在消费端维护一个短暂的重排序窗口（如 500ms），收到的事件先入缓冲队列，按 blockNumber 排序后再分发。

The Graph 索引延迟：

GraphQL Subscription 的数据源通常是 The Graph 或 SubQuery 的索引节点。索引节点从解析链上事件到写入数据库本身有 2-30 秒的延迟。这意味着即使你的 WebSocket 重连完美，补偿查询也可能拉不到"刚发生但索引还没完成"的事件。这需要在架构层面接受一个事实：基于索引层的推送存在天然延迟窗口，不能用于毫秒级时效性要求。对于需要即时确认的场景（如交易状态），应直接用 RPC 轮询作为补充。

五、总结

GraphQL Subscription 驱动链上事件流需要解决三个工程问题：WebSocket 自动重连（指数退避 + 心跳检测）、at-least-once 消息去重（eventId 集合 + LRU 策略）、离线断线补偿（游标持久化 + HTTP 查询补偿）。

游标持久化使用 localStorage 保存最后处理的 eventId，重连后从游标位置查询丢失事件。心跳检测用 WebSocket ping/pong，dead connection 超时不依赖 TCP 超时。高频场景需要评估 WebSocket 单信道瓶颈，极端性能场景考虑 UDP + 状态差量方案。