TypeGraphQL 订阅（Subscriptions）完整实战指南：@Subscription 装饰器、PubSub 主题与分布式部署

后端GraphQLAPI设计【免费下载链接】type-graphqlCreate GraphQL schema and resolvers with TypeScript, using classes and decorators!项目地址https://gitcode.com/gh_mirrors/ty/type-graphql点击查看免费下载GraphQL 提供 Query 与 Mutation 分别满足读取和写入场景但客户端往往还希望在关注的数据发生变化时被服务器主动推送更新。为此 GraphQL 定义了第三种操作——Subscription订阅。本指南以 TypeGraphQL 的官方订阅文档为主体结合仓库源码与 simple-subscriptions 示例系统讲解如何使用Subscription()装饰器声明订阅解析器、通过pubsub系统发布主题事件、按需过滤与动态主题、接入 Redis 等分布式 PubSub 以及搭建支持 WebSocket 的订阅服务端。读完本文你将能独立实现一个可扩展、可测试的 GraphQL 实时推送 API。一、订阅在 GraphQL 中的定位GraphQL 用 Query 执行读取、用 Mutation 执行写入而Subscription 用于服务器向客户端持续推送数据变更。TypeGraphQL 原生支持订阅并借助graphql-subscriptionsApollo GraphQL 团队维护这一类 PubSub 实现完成事件的发布与订阅。在 0.16.0 版本中Subscription()装饰器、PubSub()参数装饰器与buildSchema的pubSub选项构成了完整的订阅链路src/decorators/Subscription.ts负责收集订阅元数据src/schema/schema-generator.ts负责把订阅元数据转换成 GraphQL 的subscribe字段运行期需要一个 PubSub 实例默认基于EventEmitter的进程内实现。订阅解析器的整体形态与 Query/Mutation 解析器 类似但多了一层监听主题 → 接收载荷 → 转换返回值的处理因此稍显复杂。二、创建第一个订阅解析器2.1 最小订阅Subscription 装饰器订阅解析器同样是一个普通类方法只是改用Subscription()装饰器标记class SampleResolver { // ... Subscription() newNotification(): Notification { // ... } }注意仅有Subscription()而没有提供主题时订阅只是声明了返回类型真正可运行还需要配合下面的topics或subscribe选项。2.2 指定订阅主题topics需要告诉 TypeGraphQL 我们要订阅哪些主题topics。topics支持三种形态同时推荐用 TypeScript 枚举提升类型安全参见examples/simple-subscriptions/pubsub.ts中export const enum Topic的用法class SampleResolver { // ... Subscription({ topics: NOTIFICATIONS, // 单个主题 topics: [NOTIFICATIONS, ERRORS] // 或主题数组 topics: ({ args, payload, context }) args.topic // 或动态主题函数 }) newNotification(): Notification { // ... } }单个主题字符串如NOTIFICATIONS订阅所有发布到该主题的事件主题数组同时监听多个主题。从源码看多个主题会通过Repeater.merge([...topics.map(topic pubSub.subscribe(topic, topicId))])合并成同一个异步迭代器src/schema/schema-generator.ts任一主题有事件都会触发解析器动态主题函数接收{ args, payload, context }即SubscribeResolverData包含source/args/context/info依据订阅查询参数动态决定主题典型场景如客户端传入想订阅的话题名。空数组校验Subscription({ topics: [] })会在装饰器求值阶段直接抛出MissingSubscriptionTopicsErrorsrc/decorators/Subscription.ts运行时若解析后主题仍为空数组src/schema/schema-generator.ts同样会抛出该错误保证不会出现静默的无效订阅。2.3 过滤事件filter并非主题上的每个事件都应该推送给订阅者。filter选项用于决定哪些事件触发订阅函数签名接收{ payload, args, context, info }必须返回boolean或Promisebooleanclass SampleResolver { // ... Subscription({ topics: NOTIFICATIONS, filter: ({ payload, args }) args.priorities.includes(payload.priority), }) newNotification(): Notification { // ... } }上述示例只有当载荷的priority出现在订阅参数priorities中时事件才会真正派发给该订阅者。底层实现中TypeGraphQL 用pipe(pubSubIterable, filter(payload ...))将过滤逻辑串入异步迭代流src/schema/schema-generator.ts因此过滤发生在订阅者的数据流内部而非发布端。2.4 接收载荷并转换返回值Root主题被触发后订阅解析器通过Root()装饰器接收来自 pubsub 的载荷payload并在方法体内把它转换为订阅字段要求的返回形状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(), }; } }在examples/simple-subscriptions/notification.resolver.ts中可以看到真实用法normalSubscription直接把NotificationPayload展开并补上date字段subscriptionWithFilter则演示了filter: ({ payload }) payload.id % 2 0的奇偶过滤。三、发布主题事件触发订阅的 pubsub3.1 pubsub 是什么上文一直在说触发主题触发动作来自pubsub发布/订阅系统。事件可能来自数据库的外部变更也可以在我们自己的 Mutation 中触发——比如修改了某个客户端关心的资源时发布通知。假设我们有这样一个添加评论的 Mutationclass SampleResolver { // ... Mutation(returns Boolean) async addNewComment(Arg(comment) input: CommentInput) { const comment this.commentsService.createNew(input); await this.commentsRepository.save(comment); return true; } }3.2 注入 PubSub 实例PubSub()使用PubSub()参数装饰器把pubsub注入到方法参数中即可在任意地方publish主题并广播载荷给所有订阅者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); // 触发订阅主题 const payload: NotificationPayload { message: input.content }; await pubSub.publish(NOTIFICATIONS, payload); return true; } }调用pubSub.publish(NOTIFICATIONS, payload)后所有绑定到NOTIFICATIONS主题的订阅都会被触发。3.3 只注入 publishPubSub(TOPIC_NAME)为了便于测试更容易 mock/stub也可以只注入绑定到指定主题的publish方法配合PublisherTPayload类型class SampleResolver { // ... Mutation(returns Boolean) async addNewComment( Arg(comment) input: CommentInput, PubSub(NOTIFICATIONS) publish: PublisherNotificationPayload, ) { const comment this.commentsService.createNew(input); await this.commentsRepository.save(comment); // 触发订阅主题 await publish({ message: input.content }); return true; } }相比直接注入整个 PubSub 实例这种写法把发布到哪个主题的职责收拢到参数级别单测中只需 stub 一个publish函数即可无需构造完整 PubSub 引擎。至此执行addNewCommentMutation 时所有订阅了NOTIFICATIONS主题的 Subscription 都会被触发。3.4 发布端的实现细节在 0.16.0 时代的实现中publish是 PubSub 引擎的标准能力。仓库示例examples/simple-subscriptions/notification.resolver.ts展示了在 Mutation 中直接pubSub.publish(Topic.NOTIFICATIONS, payload)的完整链路而examples/simple-subscriptions/pubsub.ts中通过createPubSub{ [Topic.NOTIFICATIONS]: [NotificationPayload] }()预先声明了主题与载荷类型的映射让publish与subscribe在编译期就能得到类型检查。四、自定义 PubSub 系统从进程内到分布式4.1 默认实现的局限默认情况下TypeGraphQL 使用graphql-subscriptions中基于EventEmitter的简单PubSub。这种方案有一个显著缺陷只在 Node.js 应用为单实例单进程时才能正确工作——事件发生在进程 A进程 B 的订阅者完全收不到。4.2 接入外部存储的 PubSub 实现为了更好的可扩展性应选用由外部存储支撑的 PubSub 实现例如基于 Redis 的graphql-redis-subscriptions包。接入方式非常简单按包说明创建 PubSub 实例然后在buildSchema选项中传入即可const myRedisPubSub getConfiguredRedisPubSub(); const schema await buildSchema({ resolvers: [__dirname /**/*.resolver.ts], pubSub: myRedisPubSub, });buildSchema的pubSub选项是订阅功能的开关从源码看当存在订阅处理器而pubSub未提供时schema 生成阶段会抛出MissingPubSubErrorsrc/schema/schema-generator.ts防止生成一个无法工作的订阅 Schema。4.3 Redis 订阅的生产级示例仓库中的 redis-subscriptions 示例 演示了生产环境推荐方案。其pubsub.ts使用graphql-yoga/redis-event-target的createRedisEventTarget创建发布/订阅双客户端并配置了retryStrategy重连策略import { createRedisEventTarget } from graphql-yoga/redis-event-target; import { createPubSub } from graphql-yoga/subscription; import { Redis } from ioredis; const redisUrl process.env.REDIS_URL; if (!redisUrl) { throw new Error(REDIS_URL env variable is not defined); } 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 客户端在订阅模式下无法执行普通命令的问题重试策略retryStrategy让连接中断后按指数退避自动重连运行前提需要本地有运行中的 Redis 实例且可能需按实际连接参数修改示例代码。五、创建订阅服务端WebSocket 传输bootstrap 指南 与之前的示例都用apollo-server创建 GraphQL API 的 HTTP 端点。好消息是订阅不需要我们手动实现传输层。HTTP 无法做到真正的推送式通信订阅依赖 WebSocket而apollo-server内置了基于 WebSocket 的订阅支持开箱即用无需改动 bootstrap 配置。如果愿意也可以显式提供subscriptions配置项来定制路径与钩子// 创建 GraphQL server const server new ApolloServer({ schema, subscriptions: { path: /subscriptions, // 其他选项与钩子如 onConnect }, });完成之后我们就在/subscriptions路径上得到了一个可用的 GraphQL 订阅服务端同时保留原来的 HTTP GraphQL 服务端。六、从 0.16.0 到现代版本的订阅演进本文对应的文档版本0.16.0以graphql-subscriptionsPubSubEngine/Publisher类型为基础设施。仓库最新文档 docs/subscriptions.md 则展示了订阅 API 的演进PubSub 创建方式现代版本推荐createPubSub()来自graphql-yoga/subscription并支持用泛型声明主题与载荷类型映射createPubSub{ NOTIFICATIONS: [NotificationPayload] }()自定义 subscribe 逻辑新增subscribe选项可直接返回AsyncIterable或PromiseAsyncIterable例如对接 Prisma 订阅能力但不能与topics/filter混用动态主题 IDtopicId支持将主题限定到具体实体 ID如topicId: ({ context }) context.userId发布时以第二个参数传入 ID实现按实体隔离的事件流PubSub 接口约定任何 PubSub 系统只要满足publish(routingKey, ...args)与subscribe(routingKey, dynamicId?)两个方法即可接入见src/typings/subscriptions.ts导出的PubSub接口Apollo Server 3 之后apollo-server不再内置订阅支持需要按官方文档手动开启更现代的方案是使用graphql-yoga见 simple-subscriptions 示例 中createYoga的用法。七、端到端示例simple-subscriptions 全流程仓库的 simple-subscriptions 示例 是理解订阅全流程的最佳教材定义类型与载荷notification.type.ts声明Notification对外返回与NotificationPayload发布载荷创建 PubSub 并声明主题pubsub.ts中export const enum Topic定义主题常量createPubSub...()建立主题与载荷的类型映射编写订阅解析器notification.resolver.ts演示了无过滤订阅、filter过滤订阅、多主题订阅topics: [Topic.NOTIFICATIONS, NOTIFICATIONS_2]、基于参数动态主题topics: ({ args }) args.topic以及动态主题 IDtopicId五种订阅形态以及配套的pubSubMutation、publishToDynamicTopic、publishWithDynamicTopicId三个发布 Mutation启动服务index.ts中buildSchema({ resolvers: [NotificationResolver], pubSub })构建 Schema再用graphql-yoga的createYoga({ schema })创建同时支持 HTTP 与 WebSocket 的服务器。八、最佳实践小结用枚举管理主题把主题字符串收敛为const enum避免拼写错误同时获得类型提示过滤放在订阅端filter在订阅者的异步迭代器内部执行可按不同订阅参数各自过滤发布端只需广播全量事件发布端注入最小依赖用PubSub(TOPIC)注入绑定主题的publish便于单测 mock生产环境务必使用分布式 PubSub单进程 EventEmitter 在部署多实例时会丢失跨进程事件Redis 等外部存储是推荐方案确保buildSchema传入pubSub否则存在订阅时 schema 生成会抛出MissingPubSubError注意版本差异0.16.0 基于graphql-subscriptions现代版本基于graphql-yoga/subscription的createPubSub与PubSub接口迁移时需同步调整发布端代码与服务器配置Apollo Server 3 需手动启用订阅或改用 graphql-yoga。赞分享后端GraphQLAPI设计【免费下载链接】type-graphqlCreate GraphQL schema and resolvers with TypeScript, using classes and decorators!项目地址https://gitcode.com/gh_mirrors/ty/type-graphql点击查看免费下载相关推荐mold 项目中的 Zstandard 测试体系从 CI 分层到调试节压缩的实战验证mold 项目中的 Zstandard 测试体系从 CI 分层到调试节压缩的实战验证 本文以 mold 仓库内置的 Zstandardzstd组件测试文档后端GraphQLAPI设计TypeGraphQL 订阅Subscriptions实战指南用装饰器构建实时 GraphQL 推送TypeGraphQL 订阅Subscriptions实战指南用装饰器构建实时 GraphQL 推送 GraphQL 标准定义了三种操作类型用 Quer后端GraphQLAPI设计GraphQL Yoga 订阅实战基于 examples/subscriptions 掌握 PubSub、Repeater 与 WebSocket 订阅全流程GraphQL Yoga 订阅实战基于 examples/subscriptions 掌握 PubSub、Repeater 与 WebSocket 订阅全流程后端API设计创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考