Prisma 1 GraphQL 订阅Subscriptions完全指南实时监听数据变更的 API 实战与底层原理【免费下载链接】prisma1 Database Tools incl. ORM, Migrations and Admin UI (Postgres, MySQL MongoDB) [deprecated]项目地址: https://gitcode.com/gh_mirrors/pr/prisma1GraphQL 订阅Subscriptions是 Prisma 1 数据 API即生成的 Prisma GraphQL API中用于实时接收数据变更通知的核心能力。本指南以 docs/1.11/04-Reference/03-Prisma-API/05-Subscriptions.md 为骨架结合仓库中server/servers/subscriptions模块的 Scala 源码与测试用例系统讲解订阅触发事件、类型订阅created/updated/deleted、WebSocket 协议交互、节点与字段过滤、关系订阅的 workaround 以及多订阅组合等完整实战方案。读完你将能够通过 GraphQL Playground 或原生 WebSocket 建立订阅连接按节点、字段与变更类型精确过滤订阅事件并理解订阅事件从数据库变更到推送给客户端之间的完整链路。概述什么触发了订阅GraphQL 订阅让你在数据发生变化时实时收到通知。Prisma 的数据 API 定义了三种触发订阅的事件events一个新节点被创建CREATED一个已有节点被更新UPDATED一个已有节点被删除DELETED下面是一个典型的订阅示例每当有新的Post节点被创建时服务器推送的 payload 会包含该Post的description和imageUrl字段subscription newPosts { post(where: { mutation_in: [CREATED] }) { mutation node { description imageUrl } } }订阅使用一个专用的 WebSocket 端点websocket endpoint进行管理而不是普通的 HTTP 查询端点。可用的订阅清单如下可通过服务内的 GraphQL Playground 探索对于数据模型中的每一个对象类型object type都会自动生成一个类型订阅type subscription用于监听该类型上的数据变更目前在关系relation中连接或断开节点不会触发订阅后文会介绍一种基于UPDATED订阅的 workaround见“关系订阅”一节。你可以在一个订阅请求中组合多个订阅触发器精确控制希望收到通知的事件订阅 API 同样复用了查询Queries中提供的强大过滤系统。订阅请求如何发送订阅订阅请求可以通过以下方式发送Apollo Client使用apollo-link-ws库可以方便地发起订阅这是 GraphQL 生态中最常用的做法社区中有大量基于 React 的实时聊天类示例项目可以参照GraphQL PlaygroundPlayground 内建对订阅的支持可直接探索和运行订阅任意 WebSocket 客户端按下面的“原生 WebSockets”一节手动实现。使用 PlaygroundGraphQL Playground 可以用来探索和运行 GraphQL 订阅。它会在内部自动完成 WebSocket 握手、subscription_start发送与subscription_data接收的流程是调试订阅最快捷的方式。使用原生 WebSockets五步流程订阅通过 WebSocket 管理。完整流程分为以下五步每一步都对应着订阅协议graphql-subscriptions中一个特定的消息类型。仓库源码server/servers/subscriptions/src/main/scala/com/prisma/subscriptions/protocol/SubscriptionProtocol.scala中SubscriptionProtocolV05定义了这套协议的完整消息集其协议名即为graphql-subscriptions。1. 建立连接首先建立 WebSocket 连接并指定graphql-subscriptions协议let webSocket new WebSocket(wss://__CLUSTER__.prisma.sh/__WORKSPACE__/__SERVICE__/__STAGE__, graphql-subscriptions);其中 URL 中的__CLUSTER__、__WORKSPACE__、__SERVICE__、__STAGE__需要替换为你的实际集群地址、工作空间、服务名与阶段名。2. 发起握手监听open事件然后向服务器发送一条type为init的 JSON 消息完成握手webSocket.onopen (event) { const message { type: init } webSocket.send(JSON.stringify(message)) }在源码协议定义中这对应客户端 → 服务器的INIT消息val INIT init // Client - Server。3. 接收消息服务器会以不同的type属性返回多种消息你可以针对每种消息做出相应处理webSocket.onmessage (event) { const data JSON.parse(event.data) switch (data.type) { case init_success: { console.log(init_success, the handshake is complete) break } case init_fail: { throw { message: init_fail returned from WebSocket server, data } } case subscription_data: { console.log(subscription data has been received, data) break } case subscription_success: { console.log(subscription_success) break } case subscription_fail: { throw { message: subscription_fail returned from WebSocket server, data } } } }这些消息类型与源码中SubscriptionProtocolV05.MessageTypes的服务器 → 客户端消息一一对应init_success握手成功、init_fail握手失败、subscription_success订阅建立成功、subscription_fail订阅建立失败、subscription_data订阅数据推送。此外协议还定义了keepalive心跳消息用于保持连接活性。4. 订阅数据变更发送type为subscription_start的消息来订阅数据变更const message { id: 1, type: subscription_start, query: subscription newPosts { post(filter: { mutation_in: [CREATED] }) { mutation node { description imageUrl } } } } webSocket.send(JSON.stringify(message))之后你会收到一条subscription_success消息当数据发生变化时则会收到subscription_data消息。subscription_start消息中携带的id属性会出现在所有subscription_data消息中因此你可以在一条 WebSocket 连接上多路复用multiplex多个订阅通过id区分不同订阅的数据。源码中SubscriptionsManagerForModel.Requests.StartSubscription以id: StringOrInt唯一标识一个订阅正是对这种多路复用能力的支撑。5. 取消订阅发送type为subscription_end的消息即可取消订阅const message { id: 1, type: subscription_end } webSocket.send(JSON.stringify(message))在源码中这对应SUBSCRIPTION_END消息SubscriptionsManagerForModel收到EndSubscription请求后会把对应id的订阅从订阅集合中移除。补充仓库同时实现了graphql-ws协议SubscriptionProtocolV07即 Apollosubscriptions-transport-ws现代版本所用的connection_init/start/stop/data/complete消息体系。使用新版 Apollo 生态时订阅请求与响应将遵循这套消息语义其连接初始化、开始订阅、停止订阅的流程与本节的五步流程在概念上一一对应。类型订阅监听某个对象类型上的数据变更对于数据模型中每一个对象类型Prisma 会自动生成对应的类型订阅。以如下仅包含Post类型的数据模型为例type Post { id: ID! unique title: String! description: String }在生成的 Prisma API 中将出现一个post订阅用于在Post类型的节点被创建、更新或删除时通知你。订阅新创建的节点订阅所有被创建的节点使用where对象并设置mutation_in: [CREATED]subscription { post(where: { mutation_in: [CREATED] }) { mutation node { description imageUrl author { id } } } }payload 包含mutation此处返回CREATEDnode允许你查询被创建节点的信息以及其关联节点的信息。订阅特定的被创建节点利用与查询相同的过滤系统filter system通过where对象中的node参数进一步过滤。例如仅当某位特定用户关注follows了author时才通知你Post被创建subscription { post(where: { AND: [{ mutation_in: [CREATED] }, { node: { author: { followedBy_some: { id: cj03x3nacox6m0119755kmcm3 } } }] }) { mutation node { description imageUrl author { id } } } }订阅被删除的节点订阅所有被删除的节点设置mutation_in: [DELETED]subscription deletePost { post(where: { mutation_in: [DELETED] }) { mutation previousValues { id } } }payload 包含mutation此处返回DELETEDpreviousValues节点删除前的标量值scalar values。注意对于CREATED订阅previousValues始终为null。订阅特定的被删除节点同样可以使用node参数做精确过滤。例如仅当特定用户关注了author时才通知Post被删除subscription { post(where: { mutation_in: [DELETED] node: { author: { followedBy_some: { id: cj03x3nacox6m0119755kmcm3 } } } }) { mutation previousValues { id } } }订阅被更新的节点订阅所有被更新的节点设置mutation_in: [UPDATED]subscription { post(where: { mutation_in: [UPDATED] }) { mutation node { description imageUrl author { id } } updatedFields previousValues { description imageUrl } } }payload 包含mutation此处返回UPDATEDnode允许你查询被更新节点及其关联节点的信息updatedFields发生变更的字段列表previousValues节点更新前的标量值。注意对于CREATED和DELETED订阅updatedFields始终为null对于CREATED订阅previousValues始终为null。订阅特定字段的更新通过updatedFields_contains等过滤条件监听特定字段的更新。例如仅当Post的description字段被修改时才通知你subscription { post(where: { mutation_in: [UPDATED] updatedFields_contains: description }) { mutation node { description } updatedFields previousValues { description } } }与updatedFields_contains类似的过滤条件还有updatedFields_contains_every: [String!]当所有指定的字段都被更新时匹配updatedFields_contains_some: [String!]当部分指定字段被更新时匹配。注意updatedFields系列过滤条件不能与mutation_in: [CREATED]或mutation_in: [DELETED]同时使用从源码实现看previousValues与updatedFields在事件处理时有着明确的区分server/servers/subscriptions/src/main/scala/com/prisma/subscriptions/resolving/SubscriptionResolver.scala中handleDatabaseCreateEvent以previousValues None, updatedFields None执行查询handleDatabaseUpdateEvent会带上previousValues与changedFieldshandleDatabaseDeleteEvent则只带previousValuesupdatedFields None。这从底层印证了文档中的三条注意点。关系订阅监听关系变更的 workaround目前关系更新的订阅只能通过UPDATED订阅配合 workaround 实现。订阅关系变化你可以通过触碰touching节点来强制触发变更通知先在对应类型上添加一个dummy: String字段然后在关系状态发生变化的节点上更新该字段mutation updatePost { updatePost( where: { id: some-id } data: { dummy: dummy # do a dummy change to trigger update subscription } ) }其原理是关系本身的连接/断开不产生独立事件但关系变化通常会伴随对该节点的一次写操作通过更新一个无关紧要的dummy字段就能让节点产生一次UPDATED事件从而借道UPDATED订阅间接感知关系变化。若希望订阅直接支持关系触发器可以关注该特性在社区中的讨论进展。组合订阅一次订阅多个变更类型你可以在同一个订阅中订阅同一类型上的多种变更。订阅所有节点上的所有变更利用where对象的mutation_in参数选择要订阅的变更类型。例如同时订阅createPost、updatePost和deletePost对应的三种事件subscription { post(where: { mutation_in: [CREATED, UPDATED, DELETED] }) { mutation node { id description } updatedFields previousValues { description imageUrl } } }订阅特定节点上的所有变更使用where对象的node参数圈定需要通知的特定节点并与mutation_in组合。例如仅当特定用户关注了作者时才通知你该作者相关Post的创建、更新与删除subscription { post( where: { mutation_in: [CREATED, UPDATED, DELETED] } node: { author: { followedBy_some: { id: cj03x3nacox6m0119755kmcm3 } } } ) { mutation node { id description } updatedFields previousValues { description imageUrl } } }注意previousValues对CREATED订阅始终为nullupdatedFields对CREATED和DELETED订阅始终为null。高级订阅过滤你可以利用与查询相同的过滤系统通过where参数实现更复杂的组合。例如订阅所有CREATED和DELETED事件外加imageUrl字段被更新时的所有UPDATED事件subscription { post(where: { OR: [{ mutation_in: [CREATED, DELETED] }, { mutation_in: [UPDATED] updatedFields_contains: imageUrl }] }) { mutation node { id description } updatedFields previousValues { description imageUrl } } }注意在任何updatedFields过滤条件与CREATED或DELETED订阅同时出现时会返回错误。此外previousValues对CREATED始终为nullupdatedFields对CREATED与DELETED始终为null。订阅的底层实现从数据库事件到 WebSocket 推送理解了订阅 API 的用法后再来看仓库中server/servers/subscriptions模块的实现可以更深入地理解订阅事件的处理链路。订阅管理器按模型组织订阅server/servers/subscriptions/src/main/scala/com/prisma/subscriptions/resolving/SubscriptionsManagerForModel.scala为每一个模型维护一个独立的订阅管理器 Actor启动时preStart它会为模型的创建、更新、删除三条通道分别建立 pub/sub 订阅并对连接断开Terminated与 Schema 失效SchemaInvalidated做出处理收到StartSubscription时登记订阅收到EndSubscription时移除订阅对同一条数据库事件它会按“查询文本 变量”对订阅进行分组同一组只执行一次过滤查询再把结果广播给组内所有订阅者从而显著降低数据库与计算开销runInChunksOf(maxParallelism 10)控制并行度当最后一个订阅者断开时管理器会自动停止自身 Actorcontext.stop(self)。事件通道消息总线如何路由server/servers/subscriptions/src/main/scala/com/prisma/subscriptions/resolving/MutationChannelUtil.scala定义了事件通道的命名规则每个模型对应三条通道subscription:event:{projectId}:create{Model}、subscription:event:{projectId}:update{Model}、subscription:event:{projectId}:delete{Model}。服务器servers/api与servers/deploy等模块在数据变更时向对应通道发布事件消息订阅服务通过消息总线message-bus消费这些消息并分发给对应模型的订阅管理器管理器再根据通道名解析出变更类型Created/Updated/Deleted——这正是mutation_in语义的底层来源。事件解析35ms 延迟缓冲与 payload 组装SubscriptionResolver.scala负责把数据库事件解析为订阅响应它根据mutationType将事件 JSON 解析为DatabaseCreateEvent、DatabaseUpdateEvent或DatabaseDeleteEvent三种类型出于读写分离一致性的考虑生产环境从可能落后主库最多约 20ms 的从库读取数据因此代码中人为加入了35ms 的缓冲延迟源码注释明确提醒不要移除该延迟创建事件只携带nodeId查询当前节点更新事件携带previousValues与changedFields即updatedFields与previousValues的 payload 来源删除事件携带previousValues。这与前文“类型订阅”一节描述的 payload 语义完全一致。协议层与文档五步流程的对应SubscriptionProtocol.scala中定义了两种协议SubscriptionProtocolV05协议名graphql-subscriptions即本文“原生 WebSockets”一节所使用的协议消息类型包括init/init_success/init_fail/subscription_start/subscription_end/subscription_success/subscription_fail/subscription_data/keepalive与文档中的五步交互一一对应SubscriptionProtocolV07协议名graphql-ws更现代的订阅协议消息类型为connection_init/connection_ack/connection_error/kakeep-alive/start/stop/data/error/complete。测试印证仓库在server/servers/subscriptions/src/test/scala/com/prisma/subscriptions/specs/目录下提供了大量订阅相关测试例如SubscriptionFilterSpec.scala验证UPDATED订阅中previousValues支持枚举类型、订阅支持别名alias等过滤行为SubscriptionsProtocolV05Spec.scala与SubscriptionsProtocolV07Spec.scala分别验证两种协议的完整消息交互流程SubscriptionsManagerForModelSpec.scala验证订阅管理器的启动、订阅登记与事件分发逻辑。这些测试用例可以直接作为理解订阅 API 行为边界的参考例如订阅查询可以使用 GraphQL 别名枚举字段会出现在previousValues中mutation_in支持UPDATED等取值。小结Prisma 1 的 GraphQL 订阅围绕CREATED/UPDATED/DELETED三种事件展开每个对象类型自动生成类型订阅配合mutation_in、node过滤与updatedFields系列过滤条件可以精确订阅“某个类型的某个节点在某类变更时”的通知订阅通过专用 WebSocket 端点graphql-subscriptions或graphql-ws协议传输既可以借助 GraphQL Playground、Apollo Client 等现成工具也可以基于原生 WebSocket 按五步流程自行实现。仓库中server/servers/subscriptions模块的实现则揭示了其底层架构模型级订阅管理器、基于消息总线的create/update/delete{Model}事件通道、35ms 的一致性缓冲以及分组复用查询的优化策略。掌握这套 API 与实现原理即可在基于 Prisma 1 的应用中构建实时通知、在线协作、实时看板等数据驱动场景。【免费下载链接】prisma1 Database Tools incl. ORM, Migrations and Admin UI (Postgres, MySQL MongoDB) [deprecated]项目地址: https://gitcode.com/gh_mirrors/pr/prisma1创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考