如何从 List 中创建多个 GenServer 进程并映射存储在其中的数据？

Posted 2023-03-15

【中文标题】如何从 List 中创建多个 GenServer 进程并映射存储在其中的数据？

【英文标题】：How to create many GenServer processes from List and map data stored in them?

【发布时间】：2016-11-21 00:43:58

【问题描述】：

在这两种方法中，我都坚持如何通过给定的一组 id 或组映射进程，然后将存储的结构映射到过滤数据。

%group => [users] 实现。

我意识到组将与用户相反，因此我创建了一个使用组名称作为键的流程模块。

我担心将来会有很多用户在几个组中，所以我的问题是如何拆分当前的UserGroupServer 模块以保持由组名标识的许多分离的进程？我想通过组列表在初始化进程中保留当前模块的功能，另外我不知道如何映射每个进程以通过 user_id 获取组？

目前我在 Phoenix lib/myapp.ex 中只启动了一个进程，方法是将模块包含在子树列表中，因此我可以直接在 Channels 中调用 UserGroupServer。

defmodule UserGroupServer do
  use GenServer

  ## Client API
  def start_link(opts \\ []) do
   GenServer.start_link(__MODULE__, :ok, opts)
  end

  def update_user_groups_state(server, data) do
    groups, user_id = data
    GenServer.call(server, :clean_groups, user_id, :infinity)
    users = Enum.map(groups, fn(group) ->
      GenServer.call(server, :add_user_group, group, user_id, :infinity)
    end)
    Enum.count(Enum.uniq(List.flatten(users)))
  end

  def get_user_groups(server, user_id) do
    GenServer.call(server, :get_user_groups, user_id)
  end

  def users_count_in_gorup(server, group) do
    GenServer.call(server, :users_count_in_gorup, group)
  end

  ## Callbacks (Server API)

  def init(_) do
    :ok, Map.new
  end

  def handle_call(:clean_groups, user_id, _from, user_group_dict) do
    user_group_dict = user_group_dict
    |> Enum.map(fn(gr, users) -> gr, List.delete(users, user_id) end)
    |> Enum.into(%)
    :reply, user_group_dict, user_group_dict
  end

  def handle_call(:add_user_group, group, user_id, _from, user_group_dict) do
    user_group_dict = if Map.has_key?(user_group_dict, group) do
      Map.update!(user_group_dict, group, fn(users) -> [user_id | users] end)
    else
      Map.put(user_group_dict, group, [user_id])
    end
    :reply, Map.fetch(user_group_dict, group), user_group_dict
  end
end

测试：

defmodule MyappUserGroupServerTest do
  use ExUnit.Case, async: false

  setup do
    :ok, server_pid = UserGroupServer.start_link
    :ok, server_pid: server_pid
  end

  test "add users", context do
    c1 = UserGroupServer.update_user_groups_state(context[:server_pid], [:a, :b, :c], 1)
    assert(1 == c1)
    c2 = UserGroupServer.update_user_groups_state(context[:server_pid], [:c, :d], 2)
    assert(2 == c2)
    c3 = UserGroupServer.update_user_groups_state(context[:server_pid], [:x], 2)
    assert(1 == c3)
    c4 = UserGroupServer.update_user_groups_state(context[:server_pid], [:d], 1)
    assert(1 == c4)
    c5 = UserGroupServer.update_user_groups_state(context[:server_pid], [:d, :c], 2)
    assert(2 == c5)
  end
end

老办法%user => [groups]

Monitor 存储分配给 user_id 的组列表。如何找到给定组中的用户？我是否需要创建单独的进程来处理组和用户 ID 之间的 m..n 关系？我应该改变什么来获取每个用户组然后映射它们？

服务器实现：

defmodule Myapp.Monitor do
  use GenServer

  def create(user_id) do
    case GenServer.whereis(ref(user_id)) do
      nil -> Myapp.Supervisor.start_child(user_id)
    end
  end

  def start_link(user_id) do
    GenServer.start_link(__MODULE__, [], name: ref(user_id))
  end

  def set_groups(user_pid, groups) do
    try_call user_pid, :set_groups, groups
  end

  def handle_call(:set_groups, groups, _from, state) do
     :reply, groups, groups  # reset user groups on each set_group call.
  end

  defp ref(user_id) do
    :global, :user, user_id
  end

  defp try_call(user_id, call_function) do
    case GenServer.whereis(ref(user_id)) do
      nil -> :error, :invalid_user
      user_pid -> GenServer.call(user_pid, call_function)
    end
  end
end

主管：

defmodule Myapp.Supervisor do
  use Supervisor

  def start_link do
    Supervisor.start_link(__MODULE__, :ok, name: __MODULE__)
  end

  def start_child(user_id) do
    Supervisor.start_child(__MODULE__, [user_id])
  end

  def init(:ok) do
    supervise([worker(Myapp.Monitor, [], restart: :temporary)], strategy: :simple_one_for_one)
  end
end

例子：

Monitor.create(5)
Monitor.set_groups(5, ['a', 'b', 'c'])
Monitor.create(6)
Monitor.set_groups(6, ['a', 'b'])
Monitor.set_groups(6, ['a', 'c'])

# Monitor.users_in_gorup('a') # -> 2
# Monitor.users_in_gorup('b') # -> 1
# Monitor.users_in_gorup('c') # -> 2

# or eventually more desired:
# Monitor.unique_users_in_groups(['a', 'b', 'c']) # -> 2
# or return in set_groups unique_users_in_groups result

【参考方案1】：

在跳转到进程和 gen_servers 之前，您总是需要考虑数据结构。

您将如何添加数据？多常？你要怎么查询呢？多久一次？

在您的示例中，您提到了三个操作：

为用户设置组（重置所有以前设置的组） 返回组中的所有用户 返回一组组中的唯一用户

使用 Elixir 中最基本的类型（列表和地图），您可以通过两种方式排列数据：

映射其中键是用户，值是组列表 (%user => [groups]) 或反过来 (%group => [users])

对于这两种实现，您可以评估操作的速度。对于%user => [groups]：

为用户设置组是O(1)（只需更新地图中的键） 返回O(n*m) 组中的所有用户，其中n 是用户数，m 是组数（对于所有n 用户，您需要通过扫描潜在的@987654330 来检查它是否在组中@组名） 组内唯一用户同上+排序去重

对于%group => [users]的实现：

为用户设置组是O(n*m)（如果用户在那里，您需要扫描所有组，删除它然后只为新组设置）如果设置组只将用户添加到新组而不先删除它，它只会及时添加用户，与输入中的组数成正比（不是所有组） 返回群组O(1)中的所有用户 - 只需查询地图 与查询中的组数成正比 + 排序和去重

这表明，如果您的监视器更新迅速并且查询频率较低，那么第一次实施会好得多。如果您不那么频繁地更新它，那么第二个会好得多，但要一直查询它。

在没有任何参与者或 gen_server 的情况下实施其中一种解决方案并且可以判断它有效之后，您可能希望将 pid 视为映射键并重写算法。您也可以考虑仅使用一个进程来存储所有数据。这也取决于您的确切问题。祝你好运！

【讨论】：

感谢您的回答，似乎对于我的用例 %user => [groups] 更合适，组几乎在每个服务器请求上都会更改。最初我只想关注users_in_gorup 功能。唯一的问题是我无法找到如何在进程中管理此问题或如何在我的监视器中映射用户引用之外的组，这是我的第一个 GenServer。

您必须遍历 PID，而不是遍历映射键。困难的部分是跟踪它们。关于在 Elixir 中发现进程有一个完整的讨论：elixirconf.eu/elixirconf2016/sasa-juric 我现在只是将所有这些状态存储在一个 gen_server 中，并且只有在它被证明是瓶颈时才将其拆分。

正如我所说，您将迭代可能涉及大量消息传递的进程。对数据进行本地计算并将整个数据结构保持在一个进程中是个好主意。这也消除了存储 PID 的问题。