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 EventHandlerT (event: T) void | Promisevoid; interface StreamConfig { url: string; // GraphQL WebSocket endpoint subscription: string; // GraphQL subscription query variables?: Recordstring, 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 EventStreamClientT extends { eventId: string } { private ws: WebSocket | null null; private state: StreamState { connected: false, lastEventId: null, retryCount: 0, lastHeartbeat: 0, }; private handler: EventHandlerT | null null; private eventIdSet: Setstring new Set(); private disconnectResolve: (() void) | null null; private config: RequiredStreamConfig; constructor(config: StreamConfig) { this.config { maxRetries: Infinity, retryBaseDelay: 1000, retryMaxDelay: 30000, heartbeatInterval: 10000, ...config, }; } /** * 连接并开始接收事件 * * 为什么手动实现 WebSocket 而不是用 graphql-ws 的 createClient * 因为 graphql-ws 的自带重连不处理重连期间丢失的事件补偿 * 我们需要在重连后查询数据库拉取离线期间的事件。 */ connect(handler: EventHandlerT): Promisevoid { this.handler handler; return this.establishConnection(); } private async establishConnection(): Promisevoid { // 从 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 Promisevoid((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 Promisevoid((resolve) { this.disconnectResolve resolve; if (this.ws) { this.ws.close(1000, Client disconnect); // 1000 正常关闭 } else { resolve(); } }); } private sendMessage(message: Recordstring, 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 useRefEventStreamClientTransferEvent | null(null); useEffect(() { const client new EventStreamClientTransferEvent({ 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 CachelocalStorage 的游标在跨设备场景失效。用户切换到手机时游标不存在补偿查询会从 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/pongdead connection 超时不依赖 TCP 超时。高频场景需要评估 WebSocket 单信道瓶颈极端性能场景考虑 UDP 状态差量方案。