Phoenix.PubSub实战教程：构建实时在线用户状态追踪功能

发布时间：2026/7/10 17:21:28

Phoenix.PubSub实战教程：构建实时在线用户状态追踪功能

【免费下载链接】phoenix_pubsubDistributed PubSub and Presence platform for the Phoenix Framework项目地址: https://gitcode.com/gh_mirrors/ph/phoenix_pubsub

Phoenix.PubSub是Phoenix框架的分布式发布订阅和在线状态追踪平台，它为构建实时应用提供了强大的基础设施支持。本教程将重点介绍如何利用Phoenix.Tracker模块快速构建高效的在线用户状态追踪功能，帮助开发者实现实时用户状态管理、在线用户列表展示等常见场景。

🚀 Phoenix.PubSub核心功能介绍

Phoenix.PubSub不仅仅是一个简单的消息广播系统，它提供了完整的分布式通信解决方案。通过Phoenix.PubSub，您可以轻松实现：

分布式消息广播：跨节点、跨进程的消息传递
实时状态追踪：跟踪用户在线状态和位置信息
集群感知：自动处理节点加入和离开
容错设计：基于CRDT（无冲突复制数据类型）的最终一致性保证

📊 Phoenix.Tracker：在线状态追踪的核心模块

Phoenix.Tracker是Phoenix.PubSub中专门用于分布式在线状态追踪的模块。它采用分片架构和心跳协议，在集群中复制在线状态信息，实现最终一致性且无冲突的分布式追踪。

快速入门：创建您的第一个Tracker

首先，在您的Phoenix应用中添加Tracker到监督树：

children = [ # 其他子进程... {MyApp.Tracker, [name: MyApp.Tracker, pubsub_server: MyApp.PubSub]} ]

接下来，实现自定义的Tracker模块：

defmodule MyApp.Tracker do use Phoenix.Tracker def start_link(opts) do opts = Keyword.merge([name: __MODULE__], opts) Phoenix.Tracker.start_link(__MODULE__, opts, opts) end def init(opts) do server = Keyword.fetch!(opts, :pubsub_server) {:ok, %{pubsub_server: server, node_name: Phoenix.PubSub.node_name(server)}} end def handle_diff(diff, state) do for {topic, {joins, leaves}} <- diff do for {key, meta} <- joins do IO.puts "用户加入: 主题 #{topic}，键 #{key}，元数据 #{inspect meta}" end for {key, meta} <- leaves do IO.puts "用户离开: 主题 #{topic}，键 #{key}，元数据 #{inspect meta}" end end {:ok, state} end end

🔧 实战：构建在线用户状态追踪系统

步骤1：用户登录时注册在线状态

当用户登录到您的应用时，使用track函数注册用户的在线状态：

defmodule MyApp.UserPresence do def user_joined(user_id, user_name, topic \\ "users:online") do pid = self() meta = %{ user_id: user_id, user_name: user_name, joined_at: DateTime.utc_now(), last_seen: DateTime.utc_now(), device: "web" } case Phoenix.Tracker.track(MyApp.Tracker, pid, topic, user_id, meta) do {:ok, _ref} -> IO.puts "用户 #{user_name} 已上线" :ok {:error, reason} -> IO.puts "追踪用户失败: #{inspect reason}" {:error, reason} end end end

步骤2：定期更新用户状态

为了保持用户在线状态的准确性，可以定期更新用户元数据：

defmodule MyApp.UserPresence do def update_user_status(user_id, status, topic \\ "users:online") do pid = self() Phoenix.Tracker.update(MyApp.Tracker, pid, topic, user_id, fn meta -> Map.merge(meta, %{ status: status, last_seen: DateTime.utc_now(), updated_at: DateTime.utc_now() }) end) end end

步骤3：获取在线用户列表

轻松获取特定主题下的所有在线用户：

defmodule MyApp.UserPresence do def list_online_users(topic \\ "users:online") do Phoenix.Tracker.list(MyApp.Tracker, topic) |> Enum.map(fn {user_id, meta} -> %{ id: user_id, name: meta.user_name, joined_at: meta.joined_at, last_seen: meta.last_seen, status: meta.status || "在线" } end) end end

步骤4：用户登出时清理状态

当用户登出或连接断开时，使用untrack函数清理状态：

defmodule MyApp.UserPresence do def user_left(user_id, topic \\ "users:online") do pid = self() case Phoenix.Tracker.untrack(MyApp.Tracker, pid, topic, user_id) do {:ok, _ref} -> IO.puts "用户 #{user_id} 已下线" :ok {:error, reason} -> IO.puts "清理用户状态失败: #{inspect reason}" {:error, reason} end end end

🎯 高级功能：实时通知与事件处理

实时用户状态变化通知

通过handle_diff回调，您可以实时响应在线状态变化：

defmodule MyApp.RealTimeTracker do use Phoenix.Tracker def handle_diff(diff, state) do for {topic, {joins, leaves}} <- diff do # 发送加入通知 for {user_id, meta} <- joins do message = %{ type: "user_joined", user_id: user_id, user_name: meta.user_name, timestamp: DateTime.utc_now() } Phoenix.PubSub.broadcast!( state.pubsub_server, topic, {:presence, message} ) end # 发送离开通知 for {user_id, meta} <- leaves do message = %{ type: "user_left", user_id: user_id, user_name: meta.user_name, timestamp: DateTime.utc_now() } Phoenix.PubSub.broadcast!( state.pubsub_server, topic, {:presence, message} ) end end {:ok, state} end end

订阅在线状态变化

客户端可以订阅在线状态变化事件：

defmodule MyApp.PresenceChannel do use Phoenix.Channel def join("presence:online", _message, socket) do # 订阅在线状态变化 Phoenix.PubSub.subscribe(MyApp.PubSub, "users:online") # 发送当前在线用户列表 online_users = MyApp.UserPresence.list_online_users() push(socket, "presence_state", %{users: online_users}) {:ok, socket} end def handle_info({:presence, message}, socket) do # 处理在线状态变化 push(socket, "presence_diff", message) {:noreply, socket} end end

⚙️ 性能优化与最佳实践

1. 合理配置Tracker参数

在启动Tracker时，可以根据应用需求调整性能参数：

{MyApp.Tracker, [ name: MyApp.Tracker, pubsub_server: MyApp.PubSub, broadcast_period: 1000, # 广播周期（毫秒） max_silent_periods: 15, # 最大静默周期数 down_period: 30_000, # 下线检测周期（毫秒） pool_size: 4 # 分片池大小 ]}

2. 元数据设计建议

保持元数据轻量：只存储必要的信息
使用原子键：提高模式匹配效率
避免存储大对象：如完整的用户对象
定期清理过期数据

3. 错误处理与监控

defmodule MyApp.PresenceMonitor do use GenServer def start_link(_) do GenServer.start_link(__MODULE__, [], name: __MODULE__) end def init(_) do # 监控Tracker进程 Process.monitor(Process.whereis(MyApp.Tracker)) {:ok, %{}} end def handle_info({:DOWN, _ref, :process, _pid, reason}, state) do # Tracker进程异常退出处理 Logger.error("Tracker进程异常退出: #{inspect reason}") # 重新启动逻辑... {:noreply, state} end end

🚨 常见问题与解决方案

问题1：状态同步延迟

现象：用户在线状态更新有延迟

解决方案：

调整broadcast_period参数减少广播间隔
使用Phoenix.Tracker.update/5立即更新关键状态
实现客户端心跳机制保持连接活跃

问题2：内存占用过高

现象：Tracker进程内存持续增长

解决方案：

定期清理长时间未更新的用户状态
限制每个主题的最大用户数
使用更紧凑的元数据结构

问题3：集群节点间状态不一致

现象：不同节点看到的在线用户列表不同

解决方案：

确保网络连接稳定
调整down_period参数以适应网络延迟
实现状态一致性检查机制

📈 实际应用场景

场景1：在线客服系统

defmodule MyApp.CustomerService do def assign_agent_to_customer(customer_id) do # 查找在线的客服人员 online_agents = Phoenix.Tracker.list(MyApp.Tracker, "agents:online") |> Enum.filter(fn {_agent_id, meta} -> meta.status == "available" end) if online_agents != [] do {agent_id, _meta} = List.first(online_agents) # 分配客服给客户 {:ok, agent_id} else {:error, :no_agents_available} end end end

场景2：多人在线协作

defmodule MyApp.CollaborationRoom do def get_room_participants(room_id) do topic = "room:#{room_id}" Phoenix.Tracker.list(MyApp.Tracker, topic) |> Enum.map(fn {user_id, meta} -> %{ id: user_id, name: meta.user_name, role: meta.role || "participant", cursor_position: meta.cursor_position, last_activity: meta.last_seen } end) end end

场景3：游戏服务器在线玩家管理

defmodule MyApp.GameServer do def match_players(game_mode) do # 根据游戏模式匹配在线玩家 available_players = Phoenix.Tracker.list(MyApp.Tracker, "players:online") |> Enum.filter(fn {_player_id, meta} -> meta.game_mode == game_mode && meta.status == "waiting" end) |> Enum.sort_by(fn {_player_id, meta} -> meta.skill_level end) # 创建匹配组 create_match_groups(available_players) end end

🔍 调试与监控技巧

1. 状态检查工具

defmodule MyApp.PresenceDebug do def debug_tracker_state(tracker_name) do # 获取所有主题 topics = get_all_topics(tracker_name) IO.puts "=== Tracker状态调试 ===" IO.puts "Tracker名称: #{inspect tracker_name}" IO.puts "活跃主题数: #{length(topics)}" for topic <- topics do users = Phoenix.Tracker.list(tracker_name, topic) IO.puts "主题: #{topic}, 用户数: #{length(users)}" for {user_id, meta} <- Enum.take(users, 3) do IO.puts " - 用户: #{user_id}, 最后活动: #{meta.last_seen}" end end end end

2. 性能监控指标

defmodule MyApp.PresenceMetrics do def collect_metrics(tracker_name) do %{ total_users: count_total_users(tracker_name), active_topics: count_active_topics(tracker_name), memory_usage: get_tracker_memory_usage(tracker_name), update_rate: calculate_update_rate(tracker_name) } end end