Prisma Subscriptions 实时数据订阅完整指南：从 WebSocket 协议到类型订阅与组合过滤

Prisma Subscriptions 实时数据订阅完整指南从 WebSocket 协议到类型订阅与组合过滤【免费下载链接】prisma1 Database Tools incl. ORM, Migrations and Admin UI (Postgres, MySQL MongoDB) [deprecated]项目地址: https://gitcode.com/gh_mirrors/pr/prisma1GraphQL Subscriptions 是 Prisma API 中用于实时接收数据变更通知的核心能力当数据模型中的节点被创建CREATED、更新UPDATED或删除DELETED时服务端会通过 WebSocket 连接主动向客户端推送事件载荷。本指南基于 docs/1.10/04-Reference/03-Prisma-API/05-Subscriptions.md 展开结合本仓库server/下订阅服务的 Scala 实现系统讲解订阅的三种事件模型、WebSocket 五步接入流程、针对单类型/单节点的精细化订阅以及利用mutation_in、updatedFields、node等参数组合出任意粒度订阅过滤的实战方案。读完本文你将能独立编写可用的订阅查询并能理解其背后的协议与分发原理。订阅机制概览GraphQL subscriptions允许你在数据发生变更时被实时通知。共有三种会触发订阅的事件类型新节点被创建CREATED既有节点被更新UPDATED既有节点被删除DELETED下面是一个示例订阅每当一个新的Post节点被创建时通知你。订阅触发时服务端发送的载荷中会包含该Post的description和imageUrlsubscription newPosts { post(where: { mutation_in: [CREATED] }) { mutation node { description imageUrl } } }订阅使用专门的 WebSocket 端点而非 HTTP 查询端点进行通信。在你的服务中可用的订阅列表如下可使用 GraphQL Playground 直接探索对于数据模型中的每一个对象类型都有一个对应的类型订阅type subscription用于监听该类型上的数据变更。目前在关系中连接或断开节点不会触发订阅下文关系订阅一节提供了可用的 workaround 方案。你可以在单次订阅请求中组合多个订阅触发器见组合订阅精确控制你希望被通知的事件。订阅 API 同样复用了查询中那套丰富的过滤系统。发起订阅请求使用 Apollo Client 时可以借助apollo-link-ws库来简化订阅的使用。你也可以直接使用 GraphQL Playground或按下文描述的方式使用任何 WebSocket 客户端。PlaygroundGraphQL Playground 可用于探索并运行 GraphQL 订阅。在 Prisma 服务中打开 Playground即可看到自动生成的Subscription类型及其订阅字段。纯 WebSocket 接入1. 建立连接订阅通过 WebSocket 管理。首先建立一个 WebSocket 连接并指定graphql-subscriptions协议let webSocket new WebSocket(wss://__CLUSTER__.prisma.sh/__WORKSPACE__/__SERVICE__/__STAGE__, graphql-subscriptions);协议名graphql-subscriptions对应仓库中的V05 协议。在 SubscriptionProtocol.scala 中可以看到SubscriptionProtocolV05.protocolName graphql-subscriptions此外服务端还实现了基于graphql-ws的V07 协议对应connection_init/start/stop/data/complete等消息类型见同文件 L21-L36。本节按文档示例以 V05 协议演示。2. 发起握手接下来需要与 WebSocket 服务器握手监听open事件然后向服务器发送一个type属性为init的 JSON 消息webSocket.onopen (event) { const message { type: init } webSocket.send(JSON.stringify(message)) }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 } } } }这些消息类型与源码一一对应SubscriptionProtocol.scala 中SubscriptionProtocolV05.MessageTypes定义了客户端到服务端的init、subscription_start、subscription_end以及服务端到客户端的init_success、init_fail、keepalive、subscription_success、subscription_fail、subscription_data。其中subscription_data由SubscriptionData(id, payload)构成id用于标识是哪条订阅发来的数据L173-L175。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))你会收到一条type为subscription_success的消息。当数据发生变更时你会收到type为subscription_data的消息。你在subscription_start消息中提供的id属性会出现在所有subscription_data消息上从而允许你在一条 WebSocket 连接上**多路复用multiplex**多条订阅。5. 取消订阅要取消对数据变更的订阅发送一条type属性为subscription_end的消息const message { id: 1, type: subscription_end } webSocket.send(JSON.stringify(message))类型订阅对于数据模型中每一个可用的对象类型都会自动生成相应的订阅。例如考虑下面这个只含单个Post类型的数据模型type Post { id: ID! unique title: String! description: String }在生成的 Prisma API 中将会有一个可用的post订阅用来在特定Post节点被创建、更新或删除时通知你。从实现上看该订阅字段由 SubscriptionSchema.scala 动态构造字段名即模型名的驼峰形式camelCase(model.name)输出类型会根据mutation、updatedFields、previousValues等上下文映射出不同的订阅载荷结构。订阅创建的节点对于给定类型你可以通过生成的类型订阅来订阅所有正在被创建的节点。订阅所有被创建的节点如果你想订阅Post类型被创建的节点可以使用Post订阅在where对象中设置mutation_in: [CREATED]subscription { post(where: { mutation_in: [CREATED] }) { mutation node { description imageUrl author { id } } } }载荷包含mutation此处将返回CREATED。node允许你查询被创建节点的信息以及它的关联节点。订阅特定被创建的节点你可以使用where对象的node参数复用与查询相同的过滤系统。例如仅当某个特定用户关注follows了author时才通知我该Post被创建subscription { post(where: { AND: [{ mutation_in: [CREATED] }, { node: { author: { followedBy_some: { id: cj03x3nacox6m0119755kmcm3 } } }] }) { mutation node { description imageUrl author { id } } } }订阅删除的节点对于给定类型你可以通过生成的类型订阅来订阅所有正在被删除的节点。订阅所有被删除的节点如果你想订阅Post类型被删除的节点可以使用Post订阅在where对象中设置mutation_in: [DELETED]subscription deletePost { post(where: { mutation_in: [DELETED] }) { mutation previousValues { id } } }载荷包含mutation此处将返回DELETED。previousValues节点被删除前的标量值。注意previousValues在CREATED订阅中始终为null。订阅特定被删除的节点同样通过where对象的node参数复用查询过滤系统。例如仅当某个特定用户关注了author时才通知我该Post被删除subscription { post(where: { mutation_in: [DELETED] node: { author: { followedBy_some: { id: cj03x3nacox6m0119755kmcm3 } } } }) { mutation previousValues { id } } }订阅更新的节点对于给定类型你可以通过生成的类型订阅来订阅所有正在被更新的节点。订阅所有被更新的节点如果你想订阅Post类型被更新的节点可以使用Post订阅在where对象中设置mutation_in: [UPDATED]subscription { post(where: { mutation_in: [UPDATED] }) { mutation node { description imageUrl author { id } } updatedFields previousValues { description imageUrl } } }载荷包含mutation此处将返回UPDATED。node允许你查询被更新节点及其关联节点的信息。updatedFields发生变更的字段列表。previousValues节点更新前的标量值。注意updatedFields在CREATED与DELETED订阅中始终为nullpreviousValues在CREATED订阅中始终为null。订阅指定字段的更新你可以通过where对象的node参数复用查询过滤系统。例如仅当某个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]一起使用这些内存中的过滤语义在服务端由 QueryTransformer.scala 精确实现evaluateMutationInFilter会解析mutation_in字段并判断当前事件类型是否命中evaluateUpdatedFieldsInFilter会解析updatedFields_contains、updatedFields_contains_everyvaluesSet.subsetOf(updatedFields)与updatedFields_contains_somevaluesSet.exists(updatedFields.contains)三类条件L37-L77。关系订阅Relation subscriptions目前关系更新的订阅只能借助UPDATED订阅以 workaround 方式实现。订阅关系变更你可以通过touch节点来强制触发通知。给相关类型添加一个dummy: String字段并在该节点的关系状态发生变化时更新这个字段mutation updatePost { updatePost( where: { id: some-id } data: { dummy: dummy # do a dummy change to trigger update subscription } ) }如果你对订阅的直接关系触发感兴趣可以关注 GitHub 上的相关功能讨论。在原文档中对关系触发感兴趣的开发者曾被引导至 graphcool/feature-requests 仓库的对应 issue 参与讨论。组合订阅Combining subscriptions你可以在同一条订阅中订阅同一类型上的多个变更。订阅所有节点的所有变更通过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组合。例如仅当某个特定用户关注了 author 时才通知我该用户的帖子被创建、更新或删除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被更新时的所有UPDATEDsubscription { 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。源码视角订阅事件如何被路由与分发理解了订阅的书写方式后再从本仓库server/的实现看一遍完整链路能帮助你更准确地预估其行为与性能特征。1. 订阅 Schema 的动态生成。每个模型类型都会生成一个订阅字段字段名 模型名驼峰化如Post→post输出类型依据mutation类型动态映射删除场景下强制返回previousValues更新场景附加updatedFields见 SubscriptionSchema.scala。2. 变更事件的信道命名。数据库中的写操作会以subscription:event:projectId:create|update|deleteModelName的格式发布到消息总线模型管理器据此判断事件对应的mutationTypeCreated/Updated/Deleted见 MutationChannelUtil.scala。3. 订阅管理器与同查询优化。每个项目, 模型组合对应一个 SubscriptionsManagerForModel Actor它在启动时订阅该模型三条变更信道并将事件路由给所有mutationTypes匹配的订阅对于查询文本与变量完全相同的多条订阅只执行一次过滤查询、把结果广播给所有订阅者L117-L141。这一同查询只算一次的行为在测试 SubscriptionsManagerForModelSpec.scala 中被显式验证两条相同查询的订阅共享一次processDatabaseEventForSubscription调用invocationCount.get() should be(1)且其中一条EndSubscription后另一条仍持续收到事件。4. 订阅查询的校验与解析。在服务端接受订阅前ValidateSubscriptionQuerySpec 等测试保障了订阅查询的合法性校验事件到达后FilteredResolver 会基于事件携带的nodeId结合订阅中的node过滤条件去数据库解析出最新的节点数据。5. 双协议支持。协议层同时实现了graphql-subscriptionsV05本文示例所用与graphql-wsV07两套消息协议并配有对应的协议测试见 server/servers/subscriptions/src/test/scala/com/prisma/subscriptions/protocol 下的SubscriptionSessionProtocolV05Spec与SubscriptionSessionProtocolV07Spec因此既可以直接用裸 WebSocket 按本文流程接入也可以无缝对接 Apollo Client 等基于标准graphql-ws协议的工具链。小结Prisma Subscriptions 以类型订阅 事件枚举 过滤系统三个维度提供了高度灵活的实时数据通知能力mutation_in控制监听哪类变更node条件精确到具体节点及其关系updatedFields_*系列进一步细化为哪些字段变了才通知我三者还可以通过AND/OR自由组合。服务端基于 WebSocketgraphql-subscriptions或graphql-ws协议承载订阅会话并通过消息总线 Actor 管理器完成事件路由与同查询优化。需要进一步了解数据模型对象类型与关系的定义方式可阅读 Prisma API 概念文档查询过滤系统的完整参数可参见 Prisma API 查询文档。【免费下载链接】prisma1 Database Tools incl. ORM, Migrations and Admin UI (Postgres, MySQL MongoDB) [deprecated]项目地址: https://gitcode.com/gh_mirrors/pr/prisma1创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考