Phoenix Channels 实时通信

LiveView vs Channels：如何选择

维度	LiveView	Channels
适用场景	单用户交互界面、表单、页面	多用户广播、游戏、聊天室
消息方向	双向，但以服务端为主	完全双向，客户端可主动广播
状态管理	服务端 assigns	自定义 socket.assigns
前端	无需写 JS（除 Hook）	需要 phoenix.js 客户端
SEO	服务端渲染，SEO 友好	纯 WebSocket，不含 HTML

Channel 基础结构

# lib/blog_web/channels/room_channel.ex
defmodule BlogWeb.RoomChannel do
  use Phoenix.Channel

  # join：客户端加入频道时调用
  @impl true
  def join("room:" <> room_id, params, socket) do
    user_id = socket.assigns.user_id
    IO.puts("用户 #{user_id} 加入房间 #{room_id}")

    # 在加入后发送历史消息
    history = Blog.Chat.recent_messages(room_id)

    socket = assign(socket, room_id: room_id)
    {:ok, %{history: history}, socket}
  end

  # handle_in：接收客户端消息
  @impl true
  def handle_in("new_message", %{"body" => body}, socket) do
    user_id = socket.assigns.user_id
    room_id = socket.assigns.room_id

    msg = %{
      id:         System.unique_integer([:positive]),
      user_id:   user_id,
      body:      body,
      timestamp: DateTime.utc_now()
    }

    # 广播给频道内所有人（包括发送者）
    broadcast!(socket, "message_received", msg)

    {:noreply, socket}
  end

  def handle_in("typing", _params, socket) do
    # broadcast_from! 不发给自己
    broadcast_from!(socket, "user_typing", %{user_id: socket.assigns.user_id})
    {:noreply, socket}
  end

  # terminate：客户端离开
  @impl true
  def terminate(_reason, socket) do
    broadcast_from!(socket, "user_left", %{user_id: socket.assigns.user_id})
    :ok
  end
end

UserSocket：连接入口

# lib/blog_web/channels/user_socket.ex
defmodule BlogWeb.UserSocket do
  use Phoenix.Socket

  # 注册 Channel 路由
  channel "room:*", BlogWeb.RoomChannel
  channel "game:*", BlogWeb.GameChannel

  # connect：WebSocket 握手时验证
  @impl true
  def connect(%{"token" => token}, socket, _connect_info) do
    case Phoenix.Token.verify(BlogWeb.Endpoint, "user_auth", token, max_age: 86400) do
      {:ok, user_id} -> {:ok, assign(socket, user_id: user_id)}
      {:error, _}    -> :error
    end
  end

  @impl true
  def id(socket), do: "user_socket:#{socket.assigns.user_id}"
end

前端 JavaScript 客户端

// assets/js/app.js
import { Socket } from "phoenix"

// 建立 Socket 连接
const socket = new Socket("/socket", {
  params: { token: window.userToken }  // 认证 token
})
socket.connect()

// 加入 Channel
const channel = socket.channel("room:general", {})

channel.join()
  .receive("ok", ({ history }) => {
    history.forEach(msg => displayMessage(msg))
  })
  .receive("error", resp => {
    console.error("加入频道失败：", resp)
  })

// 发送消息
document.getElementById("send-btn")
  .addEventListener("click", () => {
    const body = document.getElementById("message-input").value
    channel.push("new_message", { body })
      .receive("ok", () => { input.value = "" })
  })

// 接收广播消息
channel.on("message_received", (msg) => {
  displayMessage(msg)
})

channel.on("user_typing", ({ user_id }) => {
  showTypingIndicator(user_id)
})

Presence：在线用户追踪

Phoenix.Presence 是分布式在线状态追踪系统，可跨多个节点同步用户的在线/离线状态：

# lib/blog_web/channels/presence.ex
defmodule BlogWeb.Presence do
  use Phoenix.Presence,
    otp_app: :blog,
    pubsub_server: Blog.PubSub
end

# 在 Channel 中使用 Presence
def join("room:" <> room_id, _params, socket) do
  user_id = socket.assigns.user_id

  # 追踪当前用户
  BlogWeb.Presence.track(socket, user_id, %{
    online_at: DateTime.utc_now(),
    name: socket.assigns.username
  })

  # 推送当前在线列表给新加入者
  presences = BlogWeb.Presence.list(socket)
  push(socket, "presence_state", presences)

  {:ok, socket}
end

# 在线列表变化时自动广播 presence_diff 事件到客户端