GraphQL进阶

订阅（Subscription）

Query 是”我要什么数据”，Mutation 是”我要改什么数据”。Subscription 是第三种操作——“有变化了通知我”。

客户端订阅一个事件，服务端有数据更新时主动推送给客户端。跟 WebSocket 的效果类似，但走的是 GraphQL 协议。

开启订阅

Apollo 驱动推荐用 graphql-ws 库：

npm install graphql-ws graphql-subscriptions

在配置中开启：

GraphQLModule.forRoot<ApolloDriverConfig>({
  driver: ApolloDriver,
  autoSchemaFile: true,
  subscriptions: {
    'graphql-ws': true,
  },
}),

定义订阅

用 @Subscription() 装饰器配合 PubSub 实现发布-订阅：

import { Resolver, Subscription } from '@nestjs/graphql';
import { PubSub } from 'graphql-subscriptions';

const pubSub = new PubSub();

@Resolver(() => Comment)
export class CommentsResolver {
  @Subscription(() => Comment)
  commentAdded() {
    return pubSub.asyncIterableIterator('commentAdded');
  }
}

asyncIterableIterator 返回一个异步迭代器，每当有新数据发布到 commentAdded 这个 topic 上，订阅就会把数据推送给客户端。

在 Mutation 中发布事件

通常订阅跟 Mutation 配合使用——客户端发起 Mutation 修改数据，同时触发订阅推送：

@Mutation(() => Comment)
async addComment(
  @Args('postId', { type: () => Int }) postId: number,
  @Args('content') content: string,
) {
  const newComment = await this.commentsService.add({ postId, content });

  // 发布事件，推送给所有订阅者
  pubSub.publish('commentAdded', { commentAdded: newComment });

  return newComment;
}

过滤订阅

不是所有事件都需要推送给所有客户端。用 filter 函数控制：

@Subscription(() => Comment, {
  filter: (payload, variables) =>
    payload.commentAdded.postId === variables.postId,
})
commentAdded(@Args('postId', { type: () => Int }) postId: number) {
  return pubSub.asyncIterableIterator('commentAdded');
}

客户端订阅时传入 postId，只有匹配该 postId 的评论才会推送过来。

PubSub 的正确用法

生产环境不要用内存版的 PubSub——多实例部署时各实例的 PubSub 不互通。应该用 Redis 等外部存储做后端：

// 注册为 provider
{
  provide: 'PUB_SUB',
  useValue: new PubSub(), // 生产环境换成 Redis 实现
}

然后在解析器里注入：

@Resolver()
export class CommentsResolver {
  constructor(@Inject('PUB_SUB') private pubSub: PubSub) {}
}

---

Federation（联邦架构）

当你的 GraphQL 服务越来越大，所有 schema 挤在一个服务里会变得很难维护。Federation 让你把一个大 schema 拆分成多个独立的微服务，每个微服务只负责一部分 schema。

架构是这样的：

• Gateway（网关）：接收客户端请求，把查询拆分分发到各个微服务，合并结果返回
• 子图服务：每个服务维护自己那部分 schema

创建子图服务

安装依赖：

npm install @apollo/subgraph

Code First 模式下，给实体加上 @Directive 装饰器标记主键：

import { Directive, Field, ID, ObjectType } from '@nestjs/graphql';

@ObjectType()
@Directive('@key(fields: "id")')
export class User {
  @Field(() => ID)
  id: number;

  @Field()
  name: string;
}

@key(fields: "id") 告诉网关：这个类型可以通过 id 字段唯一定位。

解析器里加一个 resolveReference 方法，网关需要跨服务查数据时会调它：

@Resolver(() => User)
export class UsersResolver {
  constructor(private usersService: UsersService) {}

  @Query(() => User)
  getUser(@Args('id') id: number): User {
    return this.usersService.findById(id);
  }

  @ResolveReference()
  resolveReference(reference: { __typename: string; id: number }): User {
    return this.usersService.findById(reference.id);
  }
}

模块配置用 ApolloFederationDriver：

import { ApolloFederationDriver, ApolloFederationDriverConfig } from '@nestjs/apollo';

@Module({
  imports: [
    GraphQLModule.forRoot<ApolloFederationDriverConfig>({
      driver: ApolloFederationDriver,
      autoSchemaFile: true,
    }),
  ],
  providers: [UsersResolver],
})
export class AppModule {}

---

查询复杂度控制

GraphQL 的强大之处在于客户端可以自由组合查询。但这也意味着恶意用户可以构造一个超复杂的查询，把你的服务器打崩。

比如这种嵌套查询：

{
  authors {
    posts {
      comments {
        author {
          posts {
            comments {
              # 无限嵌套...
            }
          }
        }
      }
    }
  }
}

查询复杂度分析就是来防这个的。

安装

npm install graphql-query-complexity

创建复杂度插件

import { GraphQLSchemaHost } from '@nestjs/graphql';
import { Plugin } from '@nestjs/apollo';
import { ApolloServerPlugin, BaseContext, GraphQLRequestListener } from '@apollo/server';
import { GraphQLError } from 'graphql';
import {
  fieldExtensionsEstimator,
  getComplexity,
  simpleEstimator,
} from 'graphql-query-complexity';

@Plugin()
export class ComplexityPlugin implements ApolloServerPlugin {
  constructor(private gqlSchemaHost: GraphQLSchemaHost) {}

  async requestDidStart(): Promise<GraphQLRequestListener<BaseContext>> {
    const maxComplexity = 20;
    const { schema } = this.gqlSchemaHost;

    return {
      async didResolveOperation({ request, document }) {
        const complexity = getComplexity({
          schema,
          operationName: request.operationName,
          query: document,
          variables: request.variables,
          estimators: [
            fieldExtensionsEstimator(),
            simpleEstimator({ defaultComplexity: 1 }),
          ],
        });

        if (complexity > maxComplexity) {
          throw new GraphQLError(
            `查询太复杂了: ${complexity}，最大允许 ${maxComplexity}`,
          );
        }
      },
    };
  }
}

这个插件会在每次查询执行前计算复杂度。超过阈值就直接拒绝。

定义字段复杂度

默认每个字段复杂度是 1。你可以给某些字段设置更高的值：

@Field({ complexity: 3 })
title: string;

也可以用函数动态计算：

@Query(() => [Item], {
  complexity: (options) => options.args.count * options.childComplexity,
})
items(@Args('count') count: number) {
  return this.itemsService.getItems({ count });
}

这个查询的复杂度会随着 count 参数线性增长——查的数据越多，复杂度越高。

---

输入类型（InputType）

Mutation 的参数通常用 InputType 来定义，跟 ObjectType 类似但用 @InputType：

import { InputType, Field } from '@nestjs/graphql';

@InputType()
export class CreatePostInput {
  @Field()
  title: string;

  @Field({ nullable: true })
  content?: string;
}

在 Mutation 里使用：

@Mutation(() => Post)
createPost(@Args('input') input: CreatePostInput) {
  return this.postsService.create(input);
}

---

枚举类型

GraphQL 的枚举也用装饰器定义：

import { registerEnumType } from '@nestjs/graphql';

export enum PostStatus {
  DRAFT = 'DRAFT',
  PUBLISHED = 'PUBLISHED',
  ARCHIVED = 'ARCHIVED',
}

registerEnumType(PostStatus, {
  name: 'PostStatus',
  description: '文章状态枚举',
});

注册后就能在 @Field() 里用了：

@Field(() => PostStatus)
status: PostStatus;

小结

• Subscription 用 PubSub + @Subscription() 实现实时推送
• 生产环境 PubSub 要用 Redis 等外部存储
• Federation 把大 schema 拆成多个微服务，网关统一合并
• 查询复杂度 防止恶意嵌套查询打崩服务器
• InputType 定义 Mutation 的输入参数
• 枚举 用 registerEnumType 注册后使用