原文

在上一篇文章中, 我们了解了Elixir和Phoenix如何实现我们的应用的多客户端连接. 那篇文章中介绍了我们的应用, "Loop With Friends".

最后的也时最复杂的功能就是在后端平衡每个channel topic 或者说"jam"中的用户.从这里开始, 我们要离开Phoenix提供的便捷, 开始使用Elixir OTP 服务器来处理应用的 state.

用户体验

我们希望用户不会察觉到系统对jams 的管理. 当用户访问应用时, 服务器应当指示客户端加入哪一个jam, 并且在所用jam都满人时创建新的jam. 给用户的感觉就像是瞬间加入了一个jam, 而不需要手动地选择并加入.

此外, 用户离开的同时也有新用户加入, 我们应当让每个新用户加入当前最有趣的jam. 意思就是加入还有空位且人数最多的jam.

处理state

我们需要一个能表示所有的jam在任何时候的state的东西. 我们将使用一个Elixir Agent. Agent会为我们保存state, 并提供一个简洁的接口来管理和查询state. 让我们将它放在一个名为 JamBalancer 模块中.

"简洁的接口"很重要. 为了避免我们的模块承载过多的职能而变得很庞大, 我们要分解所需要的职能, 并将构建不同的模块来实现它们. 管理jams的state足以作为一个单独的职能, 所以我们让 JamBalancer 只做这件事.

有了jam balancer 提供的每个jam的当前的state, 我们可以确定当新用户访问app时要做什么. 让我们实现一个新的 无状态 模块, 来解决繁重的工作, 也就是回答我们关于state的问题. 让我们称这个新模块为 JamCollection .

现在为了尽量清晰地讨论后端的实现, 本篇文章中省略了很多源代码里对测试的支持. 之后我将写另一篇文章介绍如何测试后端.

设计好了之后, 让我们开始探索这些模块是如何提供jam 平衡功能的.

平衡Jams

为了确保我们的 jam balancer 可用, 我们需要让它随我们的应用一同启动. 让我们将它作为一个 worker 添加到监督树. source

# lib/loops_with_friends.ex
defmodule LoopsWithFriends do
  use Application

  # ...

  def start(_type, _args) do
    import Supervisor.Spec, warn: false

    children = [
      # ...
      worker(LoopsWithFriends.JamBalancer, []),
    ]

    opts = [
      strategy: :one_for_one,
      name: LoopsWithFriends.Supervisor
    ]
    Supervisor.start_link(children, opts)
  end

  # ...
end

当supervisor 在我们的应用加载时启动它的每个 children 时, 它会寻找我们的 balancer中的 start_link 函数. 这里我们为 Agent.start_link 提供了一层包装, 以一个新的 JamCollection 作为初始state.

# lib/loops_with_friends/jam_balancer.ex
defmodule LoopsWithFriends.JamBalancer do
  alias LoopsWithFriends.JamCollection

  @name __MODULE__

  def start_link(_opts) do
    Agent.start_link(fn -> JamCollection.new() end, name: @name)
  end
end

balancer启动后, 我们就可以开始使用它了. 当用户通过HTTP连接到我们的app时, 我们需要告诉客户端应该使用websocket加入哪一个channel topic. 我们需要向 jam balancer 提问: "现在那个jam还有空位?"

<!-- web/templates/page/index.html.eex -->
<script>
  document.addEventListener("DOMContentLoaded", function (event) {
    var elmApp = Elm.App.fullscreen({
      host: document.location.host,
      topic: "jams:<%= JamBalancer.current_jam %>"
    });
  });
</script>

在 current_jam 函数中, JamBalancer 模块会从它的 agent中取出jam 列表, 并传递给 JamCollection 模块, 调用函数 most_populated_jam_with_capacity_or_new. balancer 的职责就只是保存state, 传递给另一个模块来处理state.

# lib/loops_with_friends/jam_balancer.ex
defmodule LoopsWithFriends.JamBalancer do
  # ...

  def current_jam do
    JamCollection.most_populated_jam_with_capacity_or_new(jams)
  end

  defp jams do
    Agent.get(@name, &(&1))
  end
end

JamCollection 模块包含了真正做事的函数. 我们会看到jam 列表其实是一个映射.

在 most_populated_jam_with_capacity_or_new 函数的第一个从句, 我们首先检查了是否jam列表是空的, 如果是, 则返回一个新的 jam ID 以开始新的jam.

在第二个从句, 我们寻找人数最多且没有满员的jam. 如果所有的jam都满员了, jam_with_most_users_under_max 会返回 nil . 这时, 也会返回一个新的jam ID.

# lib/loops_with_friends/jam_collection.ex
defmodule LoopsWithFriends.JamCollection do
  @max_users 7

  def new, do: %{}

  def most_populated_jam_with_capacity_or_new(jams)
    when jams == %{}, do: uuid()

  def most_populated_jam_with_capacity_or_new(jams) do
    jam_with_most_users_under_max(jams) || uuid()
  end

  # ...
end

在这里为了节约篇幅我省略了jam_with_users_under_max 的实现, 你可以在源码中看到.

填充集合

现在, 我们有了清晰的API来向balancer 和collection 请求用户应该加入的jam. 我们所缺少的是当用户加入和离开时对jam集合的修改.

在channel 中我们要依托现有的 join 回调, 并添加一个 terminate 回调. 调用新的balancer 函数 refresh 和 remove_user 来实现我们需要的功能.

# web/channels/jam_channel.ex
defmodule LoopsWithFriends.JamChannel do
  # ...

  def join("jams:" <> jam_id, _params, socket) do
    Presence.track # ...

    JamBalancer.refresh(jam_id, Presence.list(socket))

    # ...
  end

  def terminate(msg, socket) do
    JamBalancer.remove_user(
      socket.assigns.jam_id,
      socket.assigns.user_id
    )

    msg
  end

  # ...
end

与 current_jam 类似, 我们的balancer会检索我们的 jam 集合的state, 返回改动后的值.

# lib/loops_with_friends/jam_balancer.ex
defmodule LoopsWithFriends.JamBalancer do
  # ...

  def refresh(jam_id, presence_map) do
    Agent.update @name, fn jams ->
      JamCollection.refresh(jams, jam_id, Map.keys(presence_map))
    end
  end

  def remove_user(jam_id, user_id) do
    Agent.update @name, fn jams ->
      JamCollection.remove_user(jams, jam_id, user_id)
    end
  end

  # ...
end

可在此查看完整的 JamCollection 代码.

避免Jam用户溢出

一切正常. 我们可以掌握jam 集合的状态, 还可以告诉每个到来的用户应该加入哪个jam. 只有一个问题, 关于时机.

我们将当前jam ID嵌入在客户端代码中发送给客户端. 然后客户端初始化与服务器的websocket连接. 在服务器获取当前jam ID 到客户端试图加入该jam 的间隔, 这个jam 有可能已经满员了.

举个例子, 想象两个用户同事载入应用. 此时jam里有6个用户(7人是单个jam的上限). 两个用户得到了同样的jam ID. 但是其中一个用户初始化socket连接更快一点儿. 那么另一个用户会发生什么?

他们会继续加入频道, 然而我们没有多余的位置, 所以会有坏事发生. 所以我们要加入一个检查, 阻止有人加入已经满了的channel topic.

# web/channels/jam_channel.ex
defmodule LoopsWithFriends.JamChannel do
  # ...

  def join("jams:" <> jam_id, _params, socket) do
    if JamBalancer.jam_capacity?(jam_id) do
      # ...

      {:ok,
       %{user_id: socket.assigns.user_id},
       assign(socket, :jam_id, jam_id)}
    else
      {:error,
       %{new_topic: "jams:#{@jam_balancer.current_jam}"}}
    end
  end

  # ...
end

我们询问balancer, 用户试图加入的jam是否还有空位, 如果有, 我们让他加入. 如果jam已经满了, 我们会返回一个error以及一个新的可加入的topic, 将重新加入的工作交给客户端. jam_capcity? 函数再一次在 balancer 和 collection 间传递.

# lib/loops_with_friends/jam_balancer.ex
defmodule LoopsWithFriends.JamBalancer do
  # ...

  def jam_capacity?(jam_id) do
    JamCollection.jam_capacity?(jams(), jam_id)
  end

  # ...
end

我们的服务器现在实现了我们所需的对多客户端, 以及多群组的支持. 当用户加入和离开时, jams 会被平衡并优化用户体验. 更多详细的内容请看源码.

后续

如果你发现了源码中和本文不同的地方, 那是我们为方便测试所做的修改. Elixir 认为测试与其它代码同样重要, 在下一篇里我们将讨论它.