企业官网建设流程全解析

• 后端
• GraphQL
• API设计

GraphQL 的第三种操作类型 Subscription（订阅）让服务端能够在数据变化时主动向客户端推送更新，是构建实时通知、动态列表刷新等场景的核心能力。TypeGraphQL 对订阅提供了开箱即用的支持，底层基于@graphql-yoga/subscriptions包实现。本文以 TypeGraphQL v2.0.0-rc.1 的官方文档为主线，结合仓库源码与内置示例，带你完整掌握订阅的创建、触发、动态 topic、自定义 PubSub 系统以及订阅服务器的搭建方案。

为什么需要订阅

在 GraphQL 中，Query 用于读取数据，Mutation 用于写入数据，而 Subscription 则是第三种操作类型：客户端订阅某个"主题（topic）"，当服务端数据发生变化时，服务端主动向所有订阅者推送事件。这解决了客户端需要轮询才能感知数据变化的问题，是构建聊天室、实时评论、告警通知等功能的基石。

TypeGraphQL 对订阅的原生支持，基于 The Guild 团队开发的@graphql-yoga/subscriptions直接导入了该包中的Repeater、filter、pipe等工具来构建订阅字段的底层执行逻辑。

创建订阅：@Subscription()装饰器

订阅解析器与 Query/Mutation 解析器 类似，但稍复杂一些。第一步是像往常一样定义普通类方法，只不过用@Subscription()装饰器标注：

class SampleResolver { // ... @Subscription() newNotification(): Notification { // ... } }

从 Subscription.ts 源码 可以看到，@Subscription()的核心配置项被分为两组互斥选项（通过MergeExclusive类型约束）：

• PubSubOptions：topics、topicId、filter，基于 pubsub 系统的标准订阅流程；
• SubscribeOptions：subscribe，完全自定义订阅逻辑。

两者不能混用。此外，装饰器还支持传入返回类型函数作为第一参数（如@Subscription(_returns => Notification, { topics: ... })），这在示例代码中非常常见。

指定 topics：订阅哪个主题

订阅必须提供希望订阅的 topics，可以是单个字符串、字符串数组，也可以是根据订阅参数动态生成 topic 的函数，还支持使用 TypeScript 枚举增强类型安全：

class SampleResolver { // ... @Subscription({ topics: "NOTIFICATIONS", // 单个 topic topics: ["NOTIFICATIONS", "ERRORS"], // topics 数组 topics: ({ args, context }) => args.topic, // 动态生成 topic 的函数 }) newNotification(): Notification { // ... } }

从 types.ts 的类型定义 可知，动态 topic 函数接收SubscribeResolverData（包含source、args、context、info四个字段，见 SubscribeResolverData.ts），返回string | string[]。

注意一个边界情况：当topics配置为空数组[]时，Subscription.ts 源码 会直接抛出MissingSubscriptionTopicsError，因此 topics 不能为空数组。

用 filter 过滤事件

filter选项用于决定哪些 topic 事件才会真正触发订阅。该函数应返回boolean或Promise<boolean>：

class SampleResolver { // ... @Subscription({ topics: "NOTIFICATIONS", filter: ({ payload, args }) => args.priorities.includes(payload.priority), }) newNotification(): Notification { // ... } }

filter 的类型定义 表明它接收SubscriptionHandlerData——一个包含payload、args、context、info的对象（见 SubscriptionHandlerData.ts）。其中payload是发布方发送的数据，args是客户端订阅时的参数，因此可以实现"只接收符合参数条件的事件"这类精细化过滤。

自定义 subscribe 逻辑

如果内置的 topics + filter 机制不满足需求（例如要对接 Prisma 等 ORM 自带的订阅功能），可以使用subscribe选项，它应是一个返回AsyncIterable或Promise<AsyncIterable>的函数。下面的例子来自文档，展示了如何使用 Prisma 1 的订阅能力：

class SampleResolver { // ... @Subscription({ subscribe: ({ root, args, context, info }) => { return context.prisma.$subscribe.users({ mutation_in: [args.mutationType] }); }, }) newNotification(): Notification { // ... } }

subscribe函数同样接收SubscribeResolverData（注意其参数名为source而非root，见类型定义）。这里传入的root是文档示例中的写法，实际类型字段名为source。

注意：subscribe选项与topics、filter选项不能混用。如果自定义订阅后仍需要过滤，可以借助@graphql-yoga/subscriptions包提供的filter与map辅助函数在事件流上做处理。

用 @Root() 接收 payload 并转换

订阅解析器方法本身会收到由 pubsub 系统触发的 topic 事件负载（payload），通过@Root()装饰器注入，然后可以在方法体内将其转换为最终返回给客户端的形状：

class SampleResolver { // ... @Subscription({ topics: "NOTIFICATIONS", filter: ({ payload, args }) => args.priorities.includes(payload.priority), }) newNotification( @Root() notificationPayload: NotificationPayload, @Args() args: NewNotificationsArgs, ): Notification { return { ...notificationPayload, date: new Date(), }; } }

触发订阅 topics

创建好订阅后，下一个问题是如何触发。触发可以来自外部数据源（如数据库变更），也可以来自 Mutation——当某个资源被修改、而客户端希望收到变更通知时。

假设我们有一个新增评论的 Mutation：

class SampleResolver { // ... @Mutation(returns => Boolean) async addNewComment(@Arg("comment") input: CommentInput) { const comment = this.commentsService.createNew(input); await this.commentsRepository.save(comment); return true; } }

第一步：创建 PubSub 实例

大多数情况下，直接调用@graphql-yoga/subscriptions包导出的createPubSub()函数即可。它支持通过类型参数定义每个 topic 对应的 payload 元组，从而获得编译期类型安全：

import { createPubSub } from "@graphql-yoga/subscriptions"; export const pubSub = createPubSub<{ NOTIFICATIONS: [NotificationPayload]; DYNAMIC_ID_TOPIC: [number, NotificationPayload]; }>();

第二步：在 buildSchema() 中注册

将 PubSub 实例传给buildSchema()的pubSub选项：

import { buildSchema } from "type-graphql"; import { pubSub } from "./pubsub"; const schema = await buildSchema({ resolver, pubSub, });

第三步：在 Mutation 中发布事件

最后，在 Mutation 解析器中注入 PubSub 并调用publish()触发 topic，向所有订阅者发送 payload：

import { pubSub } from "./pubsub"; class SampleResolver { // ... @Mutation(returns => Boolean) async addNewComment(@Arg("comment") input: CommentInput, @PubSub() pubSub: PubSubEngine) { const comment = this.commentsService.createNew(input); await this.commentsRepository.save(comment); // Trigger subscriptions topics const payload: NotificationPayload = { message: input.content }; pubSub.publish("NOTIFICATIONS", payload); return true; } }

完成这三步后，所有订阅了NOTIFICATIONStopic 的订阅，都会在执行addNewCommentMutation 时被触发。

仓库示例中的完整闭环

在 simple-subscriptions 示例 中，可以看到一条完整链路：pubsub.ts 通过createPubSub创建实例并用枚举Topic定义常量；notification.resolver.ts 的pubSubMutation发布事件；normalSubscription订阅后经@Root()取出NotificationPayload，组装出带date的Notification返回。subscriptionWithFilter则演示了filter的用法：payload.id % 2 === 0，只接受偶数 id 的事件。

带动态 ID 的 topic（Dynamic Topic ID）

该特性灵感同样来自底层使用的@graphql-yoga/subscriptions。有些场景下，你只希望为某个特定实体（如某个用户或商品）收发事件——此时可以用动态 topic ID 把 topic 限定到特定标识符上：

@Resolver() class NotificationResolver { @Subscription({ topics: "NOTIFICATIONS", topicId: ({ context }) => context.userId, }) newNotification(@Root() { message }: NotificationPayload): Notification { return { message, date: new Date() }; } }

发布时，需要把 topic id 作为publish()的第二个参数传入：

pubSub.publish("NOTIFICATIONS", userId, { id, message });

仓库的 simple-subscriptions 示例 中，topicId函数从订阅参数args.topicId中取值，而publishWithDynamicTopicIdMutation 则以pubSub.publish(Topic.DYNAMIC_ID_TOPIC, topicId, payload)形式发布，两处标识保持一致。

注意：动态 topic ID 需要 pubsub 系统本身的支持。如果使用的不是@graphql-yoga/subscriptions的createPubSub()，publish()的第二个参数可能被当作 payload 而不是动态 topic id，需要提前确认所选实现的语义。

使用自定义 PubSub 系统

TypeGraphQL 虽然在内部使用@graphql-yoga/subscriptions处理订阅，但并不强制要求使用它的 PubSub 实现。任何满足导出的PubSub接口的 pubsub 系统都可以接入，只要具备正确的.subscribe()与.publish()方法。

从 src/typings/subscriptions.ts 的 PubSub 接口 可以看到，TypeGraphQL 只要求实现者提供两个方法：

• publish(routingKey: string, ...args: unknown[]): void——发布事件，支持可变参数（这也是动态 topic ID 能作为第二参数传入的原因）；
• subscribe(routingKey: string, dynamicId?: unknown): AsyncIterable<unknown>——返回一个异步可迭代对象作为事件流。

这一点在生产环境尤其重要：内存事件发射器（in-memory event emitter）无法跨进程工作，多实例部署时需要改用分布式 pubsub（例如基于 Redis 的实现），确保所有实例共享同一事件通道。

生产级示例：Redis 分布式 PubSub

仓库提供了基于 Redis 的完整示例。其 pubsub.ts 通过@graphql-yoga/redis-event-target的createRedisEventTarget创建事件目标，配置一对 ioredis 客户端（一个用于 publish、一个用于 subscribe），并设置了retryStrategy重连策略：

export const pubSub = createPubSub<{ [Topic.NEW_COMMENT]: [NewCommentPayload]; }>({ eventTarget: createRedisEventTarget({ publishClient: new Redis(redisUrl, { retryStrategy: times => Math.max(times * 100, 3000), }), subscribeClient: new Redis(redisUrl, { retryStrategy: times => Math.max(times * 100, 3000), }), }), });

示例的REDIS_URL通过环境变量读取，运行前需要启动 Redis 实例，并可能要根据实际情况修改连接参数。

recipe.resolver.ts 展示了经典的"评论订阅"业务：addNewCommentMutation 在保存评论后publish到NEW_COMMENTtopic；newComments订阅用filter按recipeId过滤，确保客户端只收到自己关注菜谱的新评论。由于 Redis 跨进程传输会对 payload 做序列化，示例在订阅端用new Date(newComment.dateString)恢复Date对象（见代码注释中的说明）。

创建订阅服务器（Subscription Server）

文档此前所有示例（包括 bootstrap 指南）都使用 apollo-server 创建 GraphQL HTTP 端点。但从 Apollo Server 3 开始，内置的 "batteries-included"apollo-server包不再支持订阅，需要按官方文档指引手动开启订阅能力。

如果不想处理 Apollo Server 的订阅配置，可以直接使用graphql-yoga——仓库内置示例 simple-subscriptions/index.ts 展示了极简的订阅服务器搭建方式：buildSchema()生成可执行 schema 并传入pubSub，随后createYoga({ schema, graphqlEndpoint: "/graphql" })创建 GraphQL Yoga 服务，最后挂载到node:http服务器并监听 4000 端口即可。整个过程无需额外配置，订阅功能开箱即用。

订阅开发速查表

小结

TypeGraphQL 的订阅能力覆盖了从"定义订阅（topics / filter / 自定义 subscribe）"、"在 Mutation 中触发事件"到"动态 topic ID 限定实体维度"、"接入自定义 PubSub（含 Redis 分布式方案）"的完整链路。核心要点可归纳为：

1. @Subscription()配置中，topics+filter与subscribe是互斥的两条路径；
2. 事件触发统一走pubSub.publish(topic, payload)，而publish()的可变参数设计天然支持动态 topic ID；
3. 接入自定义 PubSub 只需实现publish与subscribe两个方法，生产环境推荐基于 Redis 的分布式实现；
4. 服务器层面，若使用 Apollo Server 3+ 需手动开启订阅支持，或直接选用 graphql-yoga 获得开箱即用的订阅体验。

想动手验证，可以直接参考仓库中的 simple-subscriptions（内存版）与 redis-subscriptions（Redis 分布式版）两个示例，后者运行前需准备好 Redis 实例。

• 后端
• GraphQL
• API设计