phoenix-api-channels

Phoenix API、Channels 和 Presence（Elixir/BEAM）

Phoenix 擅长构建 REST/JSON API 和 WebSocket Channels，所需样板代码极少，并借助 BEAM 实现容错、轻量级进程以及受监督的 PubSub/Presence。

核心支柱

• 使用控制器构建 JSON API，并通过 plugs、pipelines 和版本控制进行管理。
• Context 负责数据（Ecto schemas + 查询），并向控制器/Channels 提供精简的 API。
• 使用 Channels + PubSub 扇出实时更新；使用 Presence 跟踪用户/设备。
• 通过 plugs 进行身份验证（浏览器使用 session/cookie，API 使用 token/Bearer），并使用签名参数。

---

项目设置

mix phx.new my_api --no-html --no-live
cd my_api
mix deps.get
mix ecto.create
mix phx.server

关键文件：

• lib/my_api_web/endpoint.ex — plugs、sockets、埋点
• lib/my_api_web/router.ex — pipelines、scopes、版本控制、sockets
• lib/my_api_web/controllers/* — REST/JSON 控制器
• lib/my_api/* — contexts + Ecto schemas（负责数据逻辑）
• lib/my_api_web/channels/* — Channel 模块

---

路由和 Pipelines

分开设置浏览器和 API 的 pipelines；使用 scopes 对 API 进行版本控制。

defmodule MyApiWeb.Router do
  use MyApiWeb, :router

  pipeline :api do
    plug :accepts, ["json"]
    plug :fetch_session
    plug :protect_from_forgery
    plug MyApiWeb.Plugs.RequireAuth
  end

  scope "/api", MyApiWeb do
    pipe_through :api

    scope "/v1", V1, as: :v1 do
      resources "/users", UserController, except: [:new, :edit]
      post "/sessions", SessionController, :create
    end
  end

  socket "/socket", MyApiWeb.UserSocket,
    websocket: [connect_info: [:peer_data, :x_headers]],
    longpoll: false
end

提示

• 保持 pipelines 简短；将身份验证/防护逻辑放入 plugs。
• 为 Channels 暴露 socket "/socket"；按需限制传输方式。

---

控制器和 Plugs

控制器应保持精简；逻辑由 contexts 负责。

defmodule MyApiWeb.V1.UserController do
  use MyApiWeb, :controller
  alias MyApi.Accounts

  action_fallback MyApiWeb.FallbackController

  def index(conn, _params) do
    users = Accounts.list_users()
    render(conn, :index, users: users)
  end

  def create(conn, params) do
    with {:ok, user} <- Accounts.register_user(params) do
      conn
      |> put_status(:created)
      |> put_resp_header("location", ~p\"/api/v1/users/#{user.id}\")
      |> render(:show, user: user)
    end
  end
end

FallbackController 集中处理错误转换（{:error, :not_found} → 404 JSON）。

Plugs

• RequireAuth 验证 bearer/session tokens，并设置 current_user。
• 在 pipelines 中使用类似 plug :scrub_params 的转换，而不是在控制器中处理。
• 避免在 plugs 中执行繁重操作；它们会在每个请求中运行。

---

Contexts 和数据（Ecto）

Contexts 只向控制器/Channels 暴露所需内容。

defmodule MyApi.Accounts do
  import Ecto.Query, warn: false
  alias MyApi.{Repo, Accounts.User}

  def list_users, do: Repo.all(User)
  def get_user!(id), do: Repo.get!(User, id)

  def register_user(attrs) do
    %User{}
    |> User.registration_changeset(attrs)
    |> Repo.insert()
  end
end

指南

• Schema 模块应避免依赖控制器相关知识。
• 在 changeset 中进行验证；使用 Ecto.Multi 执行多步骤操作。
• 对于大型列表，优先使用分页辅助工具（Scrivener、Flop）。

---

Channels、PubSub 和 Presence

Channel 模块示例：

defmodule MyApiWeb.RoomChannel do
  use Phoenix.Channel
  alias Phoenix.Presence

  def join("room:" <> room_id, _payload, socket) do
    send(self(), :after_join)
    {:ok, assign(socket, :room_id, room_id)}
  end

  def handle_info(:after_join, socket) do
    Presence.track(socket, socket.assigns.user_id, %{online_at: System.system_time(:second)})
    push(socket, "presence_state", Presence.list(socket))
    {:noreply, socket}
  end

  def handle_in("message:new", %{"body" => body}, socket) do
    broadcast!(socket, "message:new", %{user_id: socket.assigns.user_id, body: body})
    {:noreply, socket}
  end
end

从 contexts 使用 PubSub

def create_order(attrs) do
  with {:ok, order} <- %Order{} |> Order.changeset(attrs) |> Repo.insert() do
    Phoenix.PubSub.broadcast(MyApi.PubSub, "orders", {:order_created, order})
    {:ok, order}
  end
end

最佳实践

• 在加入 topics 之前，于 UserSocket.connect/3 中进行授权。
• 限制负载大小；验证传入事件。
• 使用 topic 分区支持多租户（"tenant:" <> tenant_id <> ":room:" <> room_id）。

---

身份验证模式

• API tokens：接受 authorization: Bearer <token>；在 plug 中验证，并设置 current_user。
• 签名参数：使用 Phoenix.Token.sign/verify 处理短期有效的 join 参数。
• 速率限制：使用 plugs + ETS/Cachex，或反向代理（NGINX/Cloudflare）。
• CORS：在 Endpoint 中使用 cors_plug 进行配置。

---

测试

使用生成的辅助工具：

defmodule MyApiWeb.UserControllerTest do
  use MyApiWeb.ConnCase, async: true

  test "lists users", %{conn: conn} do
    conn = get(conn, ~p\"/api/v1/users\")
    assert json_response(conn, 200)["data"] == []
  end
end

Channel 测试：

defmodule MyApiWeb.RoomChannelTest do
  use MyApiWeb.ChannelCase, async: true

  test "broadcasts messages" do
    {:ok, _, socket} = connect(MyApiWeb.UserSocket, %{"token" => "abc"})
    {:ok, _, socket} = subscribe_and_join(socket, "room:123", %{})
    ref = push(socket, "message:new", %{"body" => "hi"})
    assert_reply ref, :ok
    assert_broadcast "message:new", %{body: "hi"}
  end
end

DataCase：为每项测试隔离数据库；使用 fixtures/factories 进行设置。

---

Telemetry、可观测性和运维

• :telemetry 提供来自 endpoint、controller、channel 和 Ecto 查询的事件；通过 OpentelemetryPhoenix 和 OpentelemetryEcto 导出。
• 使用 Plug.Telemetry 收集请求指标；添加日志元数据（request_id、user_id）。
• Releases：MIX_ENV=prod mix release；在 config/runtime.exs 中配置运行时。
• 集群：使用 libcluster + 分布式 PubSub 支持多节点 Presence。
• 无需静态资源的 API：对于仅提供 API 的应用，禁用未使用的 watchers（esbuild/tailwind）。

---

常见问题

• 控制器直接执行查询，而不是委托给 contexts。
• 未在 UserSocket.connect/3 中进行授权，导致 topic 暴露。
• 缺少 action_fallback → 错误格式不一致。
• 忘记限制事件负载；大型消息可能使 Channels 不堪重负。
• 未使用时仍启用 longpoll；禁用它以减少攻击面。

当 contexts 负责数据、控制器保持精简，并且 Channels 通过 PubSub/Presence 严格执行授权和遥测时，Phoenix API + Channels 表现出色。BEAM 负责并发和容错；重点应放在清晰的边界和实时体验上。