REST 的固定端点在高并发移动端场景下会暴露两个问题：一是客户端为拼装一个页面要发多次请求，二是后端难以在不破坏契约的前提下裁剪字段。GraphQL 把「取什么数据」的决定权交给客户端，用单一端点 + 强类型 Schema 换取请求次数与字段级精确性。Elixir 生态的对应实现是 Absinthe——它不是把 GraphQL 当作一层字符串解析器，而是把 Schema 编译成 Elixir 模块，让查询执行复用 BEAM 的进程与并发原语。
本文覆盖四件事：Schema 如何定义与编译；Resolver 怎样拿到上下文并访问数据；N+1 问题用 Dataloader 如何根治；Subscription 如何骑在 Phoenix Channels 上实现推送。最后讨论生产环境必须加上的复杂度限制与错误处理。
Schema 定义：代码优先还是 SDL 优先
Absinthe 支持两条路径。SDL 优先（Schema Definition Language）把类型写进 .graphql 文件，再用 Absinthe.Phase.Schema 加载；代码优先（code-first）直接用 Elixir 宏定义类型。代码优先更符合 Elixir 的表达习惯，类型即模块，编译期即可发现拼写错误。
defmodule MyApp.Schema do
use Absinthe.Schema
query do
field :user, :user do
arg :id, non_null(:id)
resolve &MyApp.Resolvers.User.get/3
end
field :posts, list_of(:post) do
arg :limit, :integer, default_value: 20
resolve &MyApp.Resolvers.Post.list/3
end
end
mutation do
field :create_post, :post do
arg :title, non_null(:string)
arg :body, non_null(:string)
resolve &MyApp.Resolvers.Post.create/3
end
end
end
use Absinthe.Schema 会在编译期把整个 Schema 编译成一个模块，导出的 __absinthe_type__/1、__absinthe_schema__/0 等函数供执行器使用。这也意味着 Schema 是编译产物，改动类型定义需要重新编译，而不是运行期热加载。
对象类型用 object 宏定义：
defmodule MyApp.Schema.Types do
use Absinthe.Schema.Notation
object :user do
field :id, non_null(:id)
field :name, non_null(:string)
field :email, :string
field :posts, list_of(:post), resolve: &MyApp.Resolvers.Post.by_user/3
end
object :post do
field :id, non_null(:id)
field :title, non_null(:string)
field :body, :string
field :author, non_null(:user)
field :inserted_at, non_null(:datetime)
end
end
Absinthe.Schema.Notation 是只含类型定义宏的轻量子模块，把类型与查询根分开可以让类型文件被多个 Schema 复用。默认标量包括 :id、:string、:integer、:float、:boolean、:datetime、:date、:decimal，其余需要自定义。
类型系统：非空、枚举、接口与联合
GraphQL 的类型系统比大多数人的第一印象要严格。三个修饰符决定了可空性：
| 写法 | 含义 | 客户端影响 |
|---|---|---|
:string | 可空字符串 | 可能返回 null |
non_null(:string) | 非空字符串 | 出错会向上冒泡 |
list_of(:string) | 字符串列表，元素可空 | 元素可能为 null |
list_of(non_null(:string)) | 元素非空 | 更安全的契约 |
non_null(list_of(non_null(:string))) | 列表与元素都非空 | 最强的契约 |
非空传播（null propagation） 是 GraphQL 最容易踩的坑：如果 non_null 字段的 Resolver 返回 nil 或抛错，null 会向上冒泡到最近的可空祖先，导致整个对象甚至整棵树变成 null。把不该非空的字段标成 non_null，会让一个小错误放大成整个响应失败。原则是：只在数据模型层面确实保证有值时才用 non_null。
枚举与接口：
enum :post_status do
value :draft, as: "draft"
value :published, as: "published"
value :archived, as: "archived"
end
interface :node do
field :id, non_null(:id)
resolve_type fn
%{__struct__: MyApp.User}, _ -> :user
%{__struct__: MyApp.Post}, _ -> :post
_, _ -> nil
end
end
resolve_type 是接口与联合类型必须提供的函数，它把运行期数据映射回具体的 GraphQL 类型名。返回值必须与 Schema 中定义的类型名一致，拼错只会在运行期报错。
自定义标量需要实现 parse/1 与 serialize/1：
scalar :uuid4 do
parse fn
%Absinthe.Blueprint.Input.String{value: value} ->
case Ecto.UUID.cast(value) do
{:ok, uuid} -> {:ok, uuid}
:error -> :error
end
_ -> :error
end
serialize &to_string/1
end
parse 处理来自客户端的输入，serialize 处理返回给客户端的输出。两者都不应抛异常，返回 :error 让 Absinthe 生成标准的 GraphQL 错误。
Resolver 与 Context
Resolver 是一个 (parent, args, resolution) -> term 的三元函数。第三个参数 resolution 携带了整个执行上下文：
defmodule MyApp.Resolvers.User do
alias Absinthe.Resolution
def get(_parent, %{id: id}, resolution) do
case MyApp.Accounts.get_user(id) do
nil -> {:error, :not_found}
user -> {:ok, user}
end
end
def me(_parent, _args, resolution) do
case Resolution.context(resolution)[:current_user] do
nil -> {:error, "unauthenticated"}
user -> {:ok, user}
end
end
end
Context 是 Resolver 与应用状态之间唯一的通道。在 Phoenix 里，通常用一个 Plug 把当前用户塞进 context：
defmodule MyAppWeb.Context do
@behaviour Plug
def init(opts), do: opts
def call(conn, _opts) do
context = build_context(conn)
Absinthe.Plug.put_options(conn, context: context)
end
defp build_context(conn) do
with ["Bearer " <> token] <- get_req_header(conn, "authorization"),
{:ok, claims} <- MyApp.Token.verify(token) do
%{current_user: MyApp.Accounts.get_user(claims["sub"])}
else
_ -> %{}
end
end
end
把这个 Plug 挂在 /api/graphql 之前，与既有的 JWT 鉴权与 Guardian 实践
是同一套 token 校验逻辑，只是出口从 Plug 变成了 context。
Resolver 的返回值语义需要记牢：
| 返回值 | 语义 |
|---|---|
{:ok, value} | 成功 |
{:error, reason} | 字段级错误，null 冒泡 |
{:error, message, extensions} | 带扩展信息的错误 |
nil | 等价于 {:ok, nil} |
| 裸值 | 直接作为结果，不推荐 |
{:middleware, ...} | 交给中间件继续处理 |
推荐始终返回 {:ok, _} 或 {:error, _} 元组，让错误路径显式。
中间件
中间件在 Resolver 前后插入逻辑，适合鉴权、日志、事务这类横切关注点：
defmodule MyApp.Middleware.RequireAuth do
@behaviour Absinthe.Middleware
def call(resolution, _config) do
case resolution.context[:current_user] do
nil ->
resolution
|> Absinthe.Resolution.put_result({:error, "unauthenticated"})
_user ->
resolution
end
end
end
# 使用
field :me, :user do
middleware MyApp.Middleware.RequireAuth
resolve &MyApp.Resolvers.User.me/3
end
中间件按声明顺序执行，可以在字段级、对象级或 Schema 级挂载。Absinthe.Middleware.Batch 与 Absinthe.Middleware.Async 是两个内置的并发中间件，前者用于批处理，后者把 Resolver 放到独立进程执行——这对于需要调用外部服务的字段很有用，单个字段的慢请求不会拖住整个查询。
Dataloader：消除 N+1
GraphQL 的树形查询天然会诱发 N+1：查询 posts { author { name } } 时，如果每个 post 的 author 都独立查库，就会产生 1 + N 次查询。Absinthe 的解法是 Dataloader——把同一批次内的请求合并成一次批量查询。
defmodule MyApp.Schema do
use Absinthe.Schema
import Absinthe.Schema.Notation
def context(ctx) do
loader =
Dataloader.new()
|> Dataloader.add_source(MyApp.Repo, MyApp.Dataloader.Ecto.new(MyApp.Repo))
Map.put(ctx, :loader, loader)
end
def plugins do
[Absinthe.Middleware.Dataloader | Absinthe.Plugin.defaults()]
end
end
关键点是 context/1 与 plugins/0 两个回调：前者为每个请求创建一个 loader，后者把 loader 的生命周期挂进执行流程。然后 Resolver 不再直接查库，而是声明「我需要什么」：
def by_user(%MyApp.User{id: id}, _args, %{context: %{loader: loader}}) do
loader
|> Dataloader.load(MyApp.Repo, :posts, id)
|> on_load(fn loader ->
posts = Dataloader.get(loader, MyApp.Repo, :posts, id)
{:ok, posts}
end)
end
Dataloader.load/4 只是登记需求，真正的批量执行发生在 Absinthe 执行器收集完同一层级的所有需求之后。on_load/2 注册的回调在批量结果就绪后执行，因此 Dataloader.get/4 一定命中缓存。
Ecto 源需要实现查询映射：
defmodule MyApp.Dataloader.Ecto do
def query(Post, %{ids: ids}, _repo, _args) do
import Ecto.Query
from p in Post, where: p.author_id in ^ids
end
end
Dataloader.Ecto 会按 ids 分组，把 N 个 id 合成一次 IN 查询，再按 id 把结果分发回各自的 Resolver。一个请求内的查询次数从 1 + N 降到 1 + 1。
Dataloader 也适用于外部服务：把 HTTP 批量接口包装成 source，就能把逐条调用合并成一次批量请求。需要注意 on_load 回调是同步执行的，如果批量查询本身很慢，会阻塞整个请求；这时应配合超时与降级。
Subscription：基于 Channels 的实时推送
Subscription 是 GraphQL 的第三种操作类型（前两种是 query 与 mutation），语义是「服务端主动推送」。Absinthe 的实现不自己造 WebSocket，而是复用 Phoenix Channels，把订阅主题映射到 channel topic。
defmodule MyApp.Schema do
use Absinthe.Schema
subscription do
field :post_created, :post do
config fn _args, _info ->
{:ok, topic: "posts:created"}
end
trigger :create_post, topic: fn _post -> ["posts:created"] end
end
end
end
三件事必须对齐：
config/2决定客户端订阅时进入哪个 topic，可以基于参数做过滤。trigger/2声明哪个 mutation 会触发推送，以及推到哪些 topic。topic可以是固定字符串，也可以是接收 mutation 结果的函数（实现「只推给相关用户」）。- Topic 命名 必须与前端订阅时的一致，否则静默收不到消息。
Channel 侧需要一个订阅用的 socket 与 channel 模块：
defmodule MyAppWeb.UserSocket do
use Phoenix.Socket
channel "__absinthe__:*", MyAppWeb.GraphQLChannel
def connect(%{"token" => token}, socket, _connect_info) do
case MyApp.Token.verify(token) do
{:ok, claims} -> {:ok, assign(socket, :user_id, claims["sub"])}
_ -> :error
end
end
end
defmodule MyAppWeb.GraphQLChannel do
use Absinthe.Phoenix.Channel
def join("__absinthe__:control", _payload, socket) do
{:ok, socket}
end
def handle_in("doc", %{"query" => query}, socket) do
Absinthe.Phoenix.Channel.handle_in("doc", %{"query" => query}, socket)
end
end
__absinthe__:control 是 Absinthe 约定的控制通道，客户端先 join 它，再发送 doc 消息建立具体订阅。推送的数据走 subscription:result 事件。这套机制与 Phoenix Channels 实时通信
完全同源：底层是 Phoenix.PubSub，进程间投递用 BEAM 消息，水平扩展靠 Phoenix.PubSub.PG2 或 Redis adapter 跨节点广播。
Subscription 的工程难点不在语法而在背压。慢客户端会积压消息，BEAM 的 mailbox 无上限增长最终会吃掉内存。缓解手段有三种：用 Phoenix.PubSub 的本地订阅减少跨节点广播；在 trigger 里做节流（合并高频事件）；对不可靠客户端设置最大订阅数与超时断开。
复杂度限制与生产加固
GraphQL 的灵活性是把双刃剑：客户端可以写出深度嵌套的查询把服务打垮。生产环境必须加三道闸。
第一道：查询深度限制。
defmodule MyApp.Schema do
use Absinthe.Schema
def plugins do
[Absinthe.Middleware.Dataloader | Absinthe.Plugin.defaults()]
end
def pipeline(pipeline) do
pipeline
|> Absinthe.Pipeline.insert_after(
Absinthe.Phase.Document.Validation.OperationName,
Absinthe.Phase.Document.Validation.DepthLimit,
max_depth: 12
)
end
end
第二道：复杂度分析。Absinthe.Phase.Document.Validation.Complexity 按字段数与嵌套深度估算成本，并允许为字段指定权重：
field :posts, list_of(:post) do
complexity fn _args, child_complexity -> 10 * child_complexity end
resolve &MyApp.Resolvers.Post.list/3
end
第三道：限流。复杂度分析给出的是「单次查询成本」，还需要按客户端维度做速率限制——把复杂度分数当作 token 消耗量记入限流桶，就能防止「合法但高频」的滥用。这与 GraphQL 限流与成本控制 的思路一致，Absinthe 侧只需把复杂度结果接到限流中间件即可。
其他生产要点：
- 持久化查询（persisted queries）：客户端只发送查询哈希，服务端查表还原，可同时降低带宽与攻击面。
- 超时：
Absinthe.Plug支持:timeout，超时后杀掉执行进程。Resolver 里调用外部服务必须自带超时，否则会拖垮整个查询。 - 错误信息脱敏：
{:error, reason}中的reason会直接进响应。生产环境应通过Absinthe.Resolution.put_result/2包装，避免泄露堆栈或 SQL 片段。 - 日志与追踪：把每个字段的执行耗时接入 Telemetry，参照 Erlang/Elixir 可观测性 的指标体系，慢字段一目了然。
HTTP 层由 Absinthe.Plug 提供，它是一个标准 Plug，可以挂在 Cowboy/Plug 路由里，与 Cowboy/Plug 路由方式一致：
forward "/api/graphql",
Absinthe.Plug,
schema: MyApp.Schema,
pipeline: {__MODULE__, :pipeline}
如果还需要 GraphiQL 调试界面，再挂一个 Absinthe.Plug.GraphiQL，只在非生产环境启用。
实践建议
- 优先代码优先 Schema。类型即模块，编译期就能发现字段名拼错、类型不匹配，比运行期报错便宜得多。
non_null要克制。非空字段的null会向上冒泡，一个未处理的边界值能让整个响应变成null。- 所有列表字段默认走 Dataloader。手写 Resolver 查库在树形查询下必然 N+1，Dataloader 的改造成本远低于事后排查。
- Subscription 先做背压设计。订阅是长连接，慢客户端的积压最终会变成内存问题。
- 三道闸（深度、复杂度、限流）缺一不可。缺深度限制可被一条查询打挂，缺限流可被高频查询打挂。
- Resolver 只做编排，不做业务。把领域逻辑放在 Context 模块里，Resolver 保持薄，才能被 mutation、REST、后台任务复用。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。