Subscription：实时数据推送

Subscription 的工作机制

GraphQL Subscription 基于 WebSocket 长连接实现。与 Query/Mutation 的请求-响应模式不同，Subscription 建立持久连接后，服务端在数据变化时主动推送给客户端。

Schema 定义

type Message {
  id: ID!
  content: String!
  sender: User!
  room: Room!
  createdAt: DateTime!
}

type Subscription {
  # 订阅特定房间的新消息
  messageSent(roomId: ID!): Message!

  # 订阅用户状态变化
  userOnlineStatus(userId: ID!): UserStatus!

  # 订阅所有新通知（无参数）
  notificationReceived: Notification!
}

服务器设置（Apollo Server + graphql-ws）

import { ApolloServer } from '@apollo/server';
import { expressMiddleware } from '@apollo/server/express4';
import { makeExecutableSchema } from '@graphql-tools/schema';
import { WebSocketServer } from 'ws';
import { useServer } from 'graphql-ws/lib/use/ws';
import { PubSub } from 'graphql-subscriptions';
import express from 'express';
import http from 'http';

const pubsub = new PubSub();
const MESSAGE_SENT = 'MESSAGE_SENT';

const resolvers = {
  Subscription: {
    messageSent: {
      // subscribe 返回 AsyncIterator
      subscribe: (_, { roomId }) => {
        return pubsub.asyncIterator([`MESSAGE_SENT_${roomId}`]);
      },
    },
  },
  Mutation: {
    sendMessage: async (_, { input }, { db }) => {
      const message = await db.messages.create({ data: input });
      // 触发订阅推送
      await pubsub.publish(`MESSAGE_SENT_${input.roomId}`, {
        messageSent: message,
      });
      return message;
    },
  },
};

const schema = makeExecutableSchema({ typeDefs, resolvers });
const app = express();
const httpServer = http.createServer(app);

// WebSocket 服务器（处理 Subscription）
const wsServer = new WebSocketServer({
  server: httpServer,
  path: '/graphql',
});
const serverCleanup = useServer({ schema }, wsServer);

const server = new ApolloServer({
  schema,
  plugins: [{
    async serverWillStart() {
      return {
        async drainServer() { await serverCleanup.dispose(); },
      };
    },
  }],
});

await server.start();
app.use('/graphql', expressMiddleware(server));
httpServer.listen(4000);

withFilter：过滤推送

import { withFilter } from 'graphql-subscriptions';

const resolvers = {
  Subscription: {
    messageSent: {
      // withFilter 只推送满足条件的事件
      subscribe: withFilter(
        // 订阅所有消息事件
        () => pubsub.asyncIterator([MESSAGE_SENT]),
        // 过滤：只推送目标 roomId 的消息
        (payload, variables) => {
          return payload.messageSent.roomId === variables.roomId;
        }
      ),
    },
  },
};

Redis PubSub（生产环境）

内置的 PubSub 是基于内存的，只适合单机环境。生产中需要使用 Redis PubSub 以支持多实例部署：

import { RedisPubSub } from 'graphql-redis-subscriptions';
import Redis from 'ioredis';

const pubsub = new RedisPubSub({
  publisher: new Redis({ host: 'redis', port: 6379 }),
  subscriber: new Redis({ host: 'redis', port: 6379 }),
});
// Redis PubSub 与内置 PubSub API 完全一致，无需修改 Resolver

客户端订阅（Apollo Client）

import { useSubscription, gql } from '@apollo/client';

const MESSAGE_SENT = gql`
  subscription MessageSent($roomId: ID!) {
    messageSent(roomId: $roomId) {
      id
      content
      sender { name avatar }
      createdAt
    }
  }
`;

function ChatRoom({ roomId }: { roomId: string }) {
  const [messages, setMessages] = useState([]);

  useSubscription(MESSAGE_SENT, {
    variables: { roomId },
    onData({ data }) {
      if (data.data?.messageSent) {
        setMessages(prev => [...prev, data.data!.messageSent]);
      }
    },
  });

  return <div>{messages.map(m => <Message key={m.id} {...m} />)}</div>;
}

Subscription 核心概念

Subscription 授权认证

// WebSocket 连接时也需要验证 Token
const serverCleanup = useServer(
  {
    schema,
    // onConnect 在握手时调用，验证 Token
    onConnect: async (ctx) => {
      const token = ctx.connectionParams?.authorization as string;
      const user = await getUserFromToken(token);
      if (!user) {
        throw new Error('Unauthorized');
        // 抛出错误会关闭 WebSocket 连接
      }
    },
    // 将用户注入到 context 中（每个 Subscription 请求都会调用）
    context: async (ctx) => {
      const token = ctx.connectionParams?.authorization as string;
      const user = await getUserFromToken(token);
      return { user, db, pubsub };
    },
  },
  wsServer
);

// 客户端发送 Token（Apollo Client）
const wsLink = new GraphQLWsLink(createClient({
  url: 'ws://localhost:4000/graphql',
  connectionParams: {
    // 在连接参数中携带 Token
    authorization: `Bearer ${localStorage.getItem('token')}`,
  },
}));

本章小结