📖 第14课：订阅与实时

阶段三：GraphQL

Subscription是GraphQL的第三种操作类型，它基于WebSocket实现服务端向客户端的实时数据推送。聊天消息、实时通知、股票行情、协作编辑——所有需要"推送"的场景都适合用Subscription。本课深入WebSocket协议、Subscription实现和生产级实践。

📡 实时通信方案对比

方案	方向	协议	优点	缺点
Polling	客户端→服务端	HTTP	简单	延迟高、开销大
Long Polling	客户端→服务端	HTTP	兼容性好	连接占用多
SSE	服务端→客户端	HTTP	简单、自动重连	单向、文本限
WebSocket	双向	WS	全双工、低延迟	需要额外状态管理
GraphQL Sub	服务端→客户端	WS	类型安全、声明式	依赖GraphQL生态

🔌 WebSocket与GraphQL Subscription

GraphQL-over-WebSocket协议

// GraphQL WebSocket协议（graphql-ws规范）

// 1. 客户端连接
ws://localhost:3000/graphql
// 子协议: graphql-transport-ws

// 2. 客户端初始化
→ { "type": "connection_init", "payload": { "authorization": "Bearer ..." } }
← { "type": "connection_ack", "payload": { "keepAlive": 10000 } }

// 3. 订阅请求
→ {
    "id": "sub1",
    "type": "subscribe",
    "payload": {
      "query": "subscription { onMessage(roomId: \"1\") { id body author { name } createdAt } }"
    }
  }

// 4. 服务端推送数据
← {
    "id": "sub1",
    "type": "next",
    "payload": {
      "data": {
        "onMessage": {
          "id": "42",
          "body": "你好！",
          "author": { "name": "张三" },
          "createdAt": "2025-01-15T10:30:00Z"
        }
      }
    }
  }

// 5. 取消订阅
→ { "id": "sub1", "type": "complete" }

// 6. 保活心跳
← { "type": "ping" }
→ { "type": "pong" }

🛠️ 实战：聊天室实时订阅

// subscription-server.js - 完整的GraphQL订阅服务
const express = require('express');
const http = require('http');
const { WebSocketServer } = require('ws');
const { buildSchema, parse, subscribe } = require('graphql');
const { createHandler } = require('graphql-http/lib/use/express');

const app = express();
const server = http.createServer(app);

// ===== Schema =====

const schema = buildSchema(`
  scalar DateTime

  type Message {
    id: ID!
    body: String!
    author: User!
    roomId: ID!
    createdAt: DateTime!
  }

  type User {
    id: ID!
    name: String!
    avatar: String
    online: Boolean!
  }

  type Room {
    id: ID!
    name: String!
    members: [User!]!
    memberCount: Int!
    messages(first: Int = 20): [Message!]!
  }

  type TypingIndicator {
    userId: ID!
    userName: String!
    roomId: ID!
  }

  type Query {
    me: User!
    room(id: ID!): Room
    rooms: [Room!]!
    messages(roomId: ID!, first: Int = 20): [Message!]!
  }

  type Mutation {
    sendMessage(roomId: ID!, body: String!): Message!
    joinRoom(roomId: ID!): Room!
    leaveRoom(roomId: ID!): Room!
    setTyping(roomId: ID!): Boolean!
  }

  type Subscription {
    onMessage(roomId: ID!): Message!
    onTyping(roomId: ID!): TypingIndicator!
    onUserOnline: User!
    onUserOffline: User!
  }
`);

// ===== 数据与发布系统 =====

const users = [
  { id: '1', name: '张三', avatar: '👤', online: true },
  { id: '2', name: '李四', avatar: '👩', online: true },
  { id: '3', name: '王五', avatar: '🧑', online: false },
];

const rooms = [
  { id: '1', name: '技术讨论', memberIds: ['1', '2'] },
  { id: '2', name: '闲聊灌水', memberIds: ['1', '2', '3'] },
];

const messages = [
  { id: '1', body: '大家好！', authorId: '1', roomId: '1', createdAt: '2025-01-15T10:00:00Z' },
  { id: '2', body: '欢迎欢迎！', authorId: '2', roomId: '1', createdAt: '2025-01-15T10:01:00Z' },
];

let nextMsgId = 3;

// 简易发布/订阅系统（生产环境用Redis Pub/Sub）
class PubSub {
  constructor() {
    this.subscribers = new Map();
  }

  subscribe(event, callback) {
    if (!this.subscribers.has(event)) this.subscribers.set(event, new Set());
    this.subscribers.get(event).add(callback);
    return () => this.subscribers.get(event)?.delete(callback);
  }

  publish(event, data) {
    const subs = this.subscribers.get(event);
    if (subs) subs.forEach(cb => cb(data));
  }
}

const pubsub = new PubSub();

// ===== Resolvers =====

const root = {
  me: () => users[0],
  room: ({ id }) => rooms.find(r => r.id === id),
  rooms: () => rooms,
  messages: ({ roomId, first }) => messages.filter(m => m.roomId === roomId).slice(-first),
  Message: {
    author: (msg) => users.find(u => u.id === msg.authorId),
  },
  Room: {
    members: (room) => room.memberIds.map(id => users.find(u => u.id === id)),
    memberCount: (room) => room.memberIds.length,
    messages: (room, { first }) => messages.filter(m => m.roomId === room.id).slice(-(first || 20)),
  },
  // Mutations
  sendMessage: ({ roomId, body }) => {
    const id = String(nextMsgId++);
    const msg = { id, body, authorId: '1', roomId, createdAt: new Date().toISOString() };
    messages.push(msg);
    pubsub.publish(`MESSAGE_${roomId}`, msg);
    return msg;
  },
  joinRoom: ({ roomId }) => {
    const room = rooms.find(r => r.id === roomId);
    if (room && !room.memberIds.includes('1')) room.memberIds.push('1');
    return room;
  },
  setTyping: ({ roomId }) => {
    pubsub.publish(`TYPING_${roomId}`, { userId: '1', userName: '张三', roomId });
    return true;
  },
  // Subscriptions
  onMessage: ({ roomId }) => {
    return {
      [Symbol.asyncIterator]() {
        const event = `MESSAGE_${roomId}`;
        const queue = [];
        let resolve;
        const callback = (data) => {
          if (resolve) { resolve({ value: data, done: false }); resolve = null; }
          else queue.push(data);
        };
        const unsub = pubsub.subscribe(event, callback);

        return {
          next() {
            if (queue.length > 0) return Promise.resolve({ value: queue.shift(), done: false });
            return new Promise(r => { resolve = r; });
          },
          return() { unsub(); return Promise.resolve({ done: true }); },
          throw(err) { unsub(); return Promise.reject(err); },
        };
      },
    };
  },
  onTyping: ({ roomId }) => {
    return {
      [Symbol.asyncIterator]() {
        const event = `TYPING_${roomId}`;
        const queue = [];
        let resolve;
        const callback = (data) => {
          if (resolve) { resolve({ value: data, done: false }); resolve = null; }
          else queue.push(data);
        };
        const unsub = pubsub.subscribe(event, callback);
        return {
          next() {
            if (queue.length > 0) return Promise.resolve({ value: queue.shift(), done: false });
            return new Promise(r => { resolve = r; });
          },
          return() { unsub(); return Promise.resolve({ done: true }); },
          throw(err) { unsub(); return Promise.reject(err); },
        };
      },
    };
  },
};

// ===== HTTP路由（Query/Mutation）=====

app.use(express.json());
app.all('/graphql', createHandler({ schema, rootValue: root }));

// ===== WebSocket路由（Subscription）=====

const wss = new WebSocketServer({ server, path: '/graphql' });

const wsClients = new Map();

wss.on('connection', (ws) => {
  const clientId = Date.now().toString(36);
  wsClients.set(clientId, { ws, subscriptions: new Map() });
  console.log(`🔌 客户端连接: ${clientId}`);

  ws.on('message', async (data) => {
    const msg = JSON.parse(data.toString());
    const client = wsClients.get(clientId);

    switch (msg.type) {
      case 'connection_init':
        ws.send(JSON.stringify({ type: 'connection_ack', payload: {} }));
        break;

      case 'subscribe': {
        const subId = msg.id;
        const doc = parse(msg.payload.query);
        const subField = doc.definitions[0].selectionSet.selections[0];
        const roomId = subField.arguments.find(a => a.name.value === 'roomId')?.value.value;

        const event = subField.name.value === 'onMessage' ? `MESSAGE_${roomId}` : `TYPING_${roomId}`;

        const unsub = pubsub.subscribe(event, (payload) => {
          // 用GraphQL执行结果格式化
          const fieldName = subField.name.value;
          const alias = subField.alias?.value || fieldName;
          ws.send(JSON.stringify({
            id: subId,
            type: 'next',
            payload: { data: { [alias]: payload } },
          }));
        });

        client.subscriptions.set(subId, { unsub, event });
        break;
      }

      case 'complete': {
        const sub = client.subscriptions.get(msg.id);
        if (sub) { sub.unsub(); client.subscriptions.delete(msg.id); }
        break;
      }

      case 'ping':
        ws.send(JSON.stringify({ type: 'pong' }));
        break;
    }
  });

  ws.on('close', () => {
    const client = wsClients.get(clientId);
    if (client) client.subscriptions.forEach(sub => sub.unsub());
    wsClients.delete(clientId);
  });
});

server.listen(3000, () => console.log('🔌 订阅服务运行在 ws://localhost:3000/graphql'));

测试Subscription

# 使用websocat测试WebSocket（安装: cargo install websocat）

# 1. 连接
websocat ws://localhost:3000/graphql -H "Sec-WebSocket-Protocol: graphql-transport-ws"

# 2. 发送初始化消息
{"type":"connection_init","payload":{}}

# 3. 订阅消息
{"id":"1","type":"subscribe","payload":{"query":"subscription { onMessage(roomId: \"1\") { id body author { name } } }"}}

# 4. 在另一个终端发送消息（触发订阅）
curl -s -X POST http://localhost:3000/graphql \
  -H "Content-Type: application/json" \
  -d '{"query":"mutation { sendMessage(roomId: \"1\", body: \"实时消息测试！\") { id body } }"}'

# 5. 观察WebSocket终端收到推送
# ← {"id":"1","type":"next","payload":{"data":{"onMessage":{"id":"3","body":"实时消息测试！","author":{"name":"张三"}}}}}

{"id":"1","type":"next","payload":{"data":{"onMessage":{"id":"3","body":"实时消息测试！","author":{"name":"张三"}}}}}

🏗️ 生产级Subscription架构

Redis Pub/Sub + 多实例

// 生产级发布/订阅：使用Redis跨实例通信
const Redis = require('ioredis');
const redis = new Redis();
const subRedis = new Redis(); // 专用订阅连接

class RedisPubSub {
  constructor() {
    this.localSubs = new Map();
    subRedis.on('message', (channel, message) => {
      const data = JSON.parse(message);
      const subs = this.localSubs.get(channel);
      if (subs) subs.forEach(cb => cb(data));
    });
  }

  async publish(channel, data) {
    // 发布到Redis，所有实例都能收到
    await redis.publish(channel, JSON.stringify(data));
  }

  subscribe(channel, callback) {
    if (!this.localSubs.has(channel)) {
      this.localSubs.set(channel, new Set());
      subRedis.subscribe(channel); // 首次订阅Redis频道
    }
    this.localSubs.get(channel).add(callback);
    return () => {
      this.localSubs.get(channel)?.delete(callback);
      if (this.localSubs.get(channel)?.size === 0) {
        subRedis.unsubscribe(channel);
        this.localSubs.delete(channel);
      }
    };
  }
}

📝 本课小结

GraphQL Subscription基于WebSocket实现实时数据推送
graphql-ws是标准的WebSocket子协议
Pub/Sub模式解耦发布者和订阅者
生产环境必须使用Redis Pub/Sub支持多实例
连接初始化、心跳保活、优雅关闭是可靠性关键
Subscription适合聊天、通知、协作等实时场景

💪 练习

练习1：实现通知系统

设计一个通知Subscription，当用户收到新消息、被关注、文章被评论时推送通知。使用union类型合并不同通知类型。

练习2：实现Redis Pub/Sub

将本课的内存PubSub替换为Redis实现，启动两个API实例验证跨实例推送。

🏆 本课成就：实时通信专家

✅ 掌握GraphQL Subscription协议
✅ 实现WebSocket + GraphQL订阅
✅ 理解Pub/Sub发布订阅模式
✅ 了解生产级Redis Pub/Sub架构