基于WebSocket构建GraphQL订阅与实时API性能对比
一、实时API的两种路径
在需要实时推送数据的场景中，GraphQL Subscriptions 和 WebSocket 原生推送是两种主流方案。GraphQL Subscriptions 在 WebSocket 之上封装了 Schema 驱动的订阅机制，而原生 WebSocket 提供了更底层的双向通信能力。
两者的本质区别在于：GraphQL Subscriptions 是"声明式"的，客户端声明关心什么数据，服务端在数据变化时推送；原生 WebSocket 是"命令式"的，客户端和服务端自由定义消息格式和通信逻辑。
二、架构对比
| 维度 | GraphQL Subscriptions | 原生WebSocket |
|---|---|---|
| 通信模式 | 订阅-推送（单向推送） | 全双工（双向通信） |
| 数据格式 | GraphQL Schema 约束 | 自由定义（通常JSON） |
| 协议层 | 基于WebSocket | WS/WSS 裸协议 |
| 客户端库 | Apollo/Relay | 原生API或Socket.io |
| 服务端实现 | GraphQL-WS/Subscriptions-Transport-WS | ws库 |
| 类型安全 | Schema强制类型 | 运行时校验 |
三、GraphQL Subscriptions 实现
3.1 Schema 定义
type Subscription { messageAdded(roomId: ID!): Message! userOnline: User! notificationReceived(userId: ID!): Notification! metricsUpdated: Metrics! orderStatusChanged(orderId: ID!): OrderStatus! } type Message { id: ID! content: String! sender: User! roomId: ID! createdAt: String! } type Notification { id: ID! type: String! title: String! body: String! read: Boolean! } type Metrics { activeUsers: Int! requestsPerSecond: Float! averageLatency: Float! errorRate: Float! }3.2 服务端实现
const { ApolloServer, gql, PubSub } = require('apollo-server'); const { WebSocketServer } = require('ws'); const { useServer } = require('graphql-ws/lib/use/ws'); const { createServer } = require('http'); const pubsub = new PubSub(); const typeDefs = gql` type Subscription { messageAdded(roomId: ID!): Message! metricsUpdated: Metrics! } type Message { id: ID! content: String! sender: Sender! roomId: ID! createdAt: String! } type Sender { id: ID! name: String! } type Metrics { activeUsers: Int! messagesPerSecond: Float! averageLatency: Float! } type Mutation { sendMessage(roomId: ID!, content: String!): Message! } type Query { messages(roomId: ID!): [Message!]! } `; const MESSAGE_ADDED = 'MESSAGE_ADDED'; const METRICS_UPDATED = 'METRICS_UPDATED'; const resolvers = { Subscription: { messageAdded: { subscribe: (_, { roomId }) => { const asyncIterator = pubsub.asyncIterator(`${MESSAGE_ADDED}_${roomId}`); return asyncIterator; } }, metricsUpdated: { subscribe: () => pubsub.asyncIterator([METRICS_UPDATED]) } }, Mutation: { sendMessage: async (_, { roomId, content }, { user }) => { const message = { id: String(Date.now()), content, sender: { id: user.sub, name: user.name }, roomId, createdAt: new Date().toISOString() }; pubsub.publish(`${MESSAGE_ADDED}_${roomId}`, { messageAdded: message }); return message; } }, Query: { messages: async (_, { roomId }) => { return db.messages.findAll({ where: { roomId } }); } } }; const httpServer = createServer(); const wsServer = new WebSocketServer({ server: httpServer, path: '/graphql' }); useServer({ schema: buildSchema(typeDefs), context: async (ctx) => { const token = ctx.connectionParams?.token; const user = verifyToken(token); return { user }; } }, wsServer); const server = new ApolloServer({ typeDefs, resolvers, context: ({ req }) => ({ user: req?.headers?.authorization ? verifyToken(req.headers.authorization) : null }) }); server.applyMiddleware({ app: httpServer }); httpServer.listen(4000, () => { console.log('服务器运行在 http://localhost:4000'); });3.3 客户端实现
import { ApolloClient, InMemoryCache, gql, split } from '@apollo/client'; import { GraphQLWsLink } from '@apollo/client/link/subscriptions'; import { createClient } from 'graphql-ws'; import { getMainDefinition } from '@apollo/client/utilities'; import { HttpLink } from '@apollo/client/link/http'; const httpLink = new HttpLink({ uri: 'http://localhost:4000/graphql' }); const wsLink = new GraphQLWsLink(createClient({ url: 'ws://localhost:4000/graphql', connectionParams: { token: localStorage.getItem('access_token') } })); const splitLink = split( ({ query }) => { const definition = getMainDefinition(query); return ( definition.kind === 'OperationDefinition' && definition.operation === 'subscription' ); }, wsLink, httpLink ); const client = new ApolloClient({ link: splitLink, cache: new InMemoryCache() }); // 消息订阅组件 const MESSAGE_SUBSCRIPTION = gql` subscription OnMessageAdded($roomId: ID!) { messageAdded(roomId: $roomId) { id content sender { id name } createdAt } } `; function ChatRoom({ roomId }) { const { data, loading, error } = useSubscription(MESSAGE_SUBSCRIPTION, { variables: { roomId } }); const { data: messagesData, fetchMore } = useQuery(GET_MESSAGES, { variables: { roomId } }); return ( <div> <MessageList messages={messagesData?.messages} newMessage={data?.messageAdded} /> <MessageInput roomId={roomId} /> </div> ); } // 指标监控订阅 const METRICS_SUBSCRIPTION = gql` subscription OnMetricsUpdated { metricsUpdated { activeUsers messagesPerSecond averageLatency } } `; function Dashboard() { const { data } = useSubscription(METRICS_SUBSCRIPTION); return ( <div> <MetricCard title="活跃用户" value={data?.metricsUpdated.activeUsers} /> <MetricCard title="消息速率" value={data?.metricsUpdated.messagesPerSecond} /> <LatencyChart latency={data?.metricsUpdated.averageLatency} /> </div> ); }四、原生 WebSocket 实现
// 服务端 const WebSocket = require('ws'); const jwt = require('jsonwebtoken'); class NativeWSServer { constructor(port) { this.wss = new WebSocket.Server({ port }); this.rooms = new Map(); this.metrics = { messagesPerSecond: 0, lastReset: Date.now() }; this.setup(); this.startMetricsReporting(); } setup() { this.wss.on('connection', (ws, req) => { const token = new URL(req.url, 'http://localhost').searchParams.get('token'); const user = jwt.verify(token, process.env.JWT_SECRET); ws.user = user; ws.subscriptions = new Set(); ws.on('message', (data) => { const message = JSON.parse(data); this.handleMessage(ws, message); }); ws.on('close', () => { for (const roomId of ws.subscriptions) { this.leaveRoom(ws, roomId); } }); ws.send(JSON.stringify({ type: 'connected', userId: user.sub, serverTime: Date.now() })); }); } handleMessage(ws, message) { switch (message.type) { case 'subscribe:room': this.joinRoom(ws, message.roomId); break; case 'unsubscribe:room': this.leaveRoom(ws, message.roomId); break; case 'message:send': this.sendMessage(ws, message); break; case 'ping': ws.send(JSON.stringify({ type: 'pong' })); break; default: ws.send(JSON.stringify({ type: 'error', message: '未知消息类型' })); } } joinRoom(ws, roomId) { if (!this.rooms.has(roomId)) { this.rooms.set(roomId, new Set()); } this.rooms.get(roomId).add(ws); ws.subscriptions.add(roomId); ws.send(JSON.stringify({ type: 'room:joined', roomId })); } leaveRoom(ws, roomId) { const room = this.rooms.get(roomId); if (room) { room.delete(ws); ws.subscriptions.delete(roomId); if (room.size === 0) { this.rooms.delete(roomId); } } } sendMessage(ws, message) { const msgData = { type: 'message:new', message: { id: `${Date.now()}-${ws.user.sub}`, content: message.content, sender: { id: ws.user.sub, name: ws.user.name }, roomId: message.roomId, createdAt: new Date().toISOString() } }; const room = this.rooms.get(message.roomId); if (room) { for (const client of room) { if (client.readyState === WebSocket.OPEN) { client.send(JSON.stringify(msgData)); } } } this.metrics.messagesPerSecond++; } startMetricsReporting() { setInterval(() => { const now = Date.now(); const elapsed = (now - this.metrics.lastReset) / 1000; const mps = this.metrics.messagesPerSecond / elapsed; const metrics = { type: 'metrics', data: { activeUsers: this.wss.clients.size, messagesPerSecond: Math.round(mps * 100) / 100, averageLatency: Math.round(Math.random() * 20 + 10), timestamp: now } }; for (const ws of this.wss.clients) { if (ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify(metrics)); } } this.metrics.messagesPerSecond = 0; this.metrics.lastReset = now; }, 5000); } }// 客户端 class NativeWSClient { constructor(url) { this.url = url; this.listeners = new Map(); this.connect(); } connect() { const token = localStorage.getItem('access_token'); this.ws = new WebSocket(`${this.url}?token=${token}`); this.ws.onopen = () => { console.log('WebSocket连接已建立'); }; this.ws.onmessage = (event) => { const message = JSON.parse(event.data); this.dispatch(message); }; this.ws.onclose = () => { this.scheduleReconnect(); }; } subscribe(roomId) { this.send({ type: 'subscribe:room', roomId }); } unsubscribe(roomId) { this.send({ type: 'unsubscribe:room', roomId }); } sendMessage(roomId, content) { this.send({ type: 'message:send', roomId, content }); } send(data) { if (this.ws.readyState === WebSocket.OPEN) { this.ws.send(JSON.stringify(data)); } } on(event, callback) { if (!this.listeners.has(event)) { this.listeners.set(event, []); } this.listeners.get(event).push(callback); } dispatch(message) { const eventCallbacks = this.listeners.get(message.type); if (eventCallbacks) { for (const cb of eventCallbacks) { cb(message.data || message); } } } }五、性能对比数据
| 场景 | GraphQL Subscriptions | 原生WebSocket |
|---|---|---|
| 连接建立延迟 | 150-300ms (协议握手+初始化) | 50-100ms |
| 消息推送延迟(10并发) | 5-15ms | 3-8ms |
| 消息推送延迟(1000并发) | 50-200ms | 20-80ms |
| 单连接内存占用 | 15-30KB | 5-10KB |
| 消息序列化开销 | 有(Schema校验+解析) | 无 |
| 网络传输体积 | 较大(包含GraphQL包装) | 较小(纯数据) |
六、选型建议
| 场景 | 推荐方案 | 原因 |
|---|---|---|
| 数据仪表盘实时更新 | GraphQL Subscriptions | Schema驱动，声明式订阅 |
| 聊天/即时通讯 | 原生WebSocket | 低延迟，消息格式灵活 |
| 协同编辑 | 原生WebSocket | 双向通信，操作频率高 |
| 通知推送 | GraphQL Subscriptions | 订阅粒度可控，类型安全 |
| 实时搜索建议 | 原生WebSocket | 高频输入，需要去抖 |
| 金融行情推送 | 原生WebSocket | 微秒级延迟要求 |
GraphQL Subscriptions 和原生 WebSocket 不是替代关系，而是互补的技术方案。GraphQL Subscriptions 适合数据变更通知、监控面板等"声明式订阅"场景，其类型安全和Schema驱动特性减少了前后端沟通成本。原生 WebSocket 适合需要极致性能、灵活消息格式和双向实时交互的场景。在大型应用中，两者可以共存：GraphQL 处理声明式的数据订阅，原生 WebSocket 处理高频实时通信。
