登录
目录
当前位置： 首页 > 文档资料 > Elixir 进阶之 Mix 和 OTP 入门 >

3. GenServer

优质
小牛编辑
162浏览
2023-12-01

在上一章我们用agent实现buckets。根据第一章所描述的设计，每个bucket应该被赋予名字，
从而可以这么去访问：

  1. CREATE shopping
  2. OK
  4. PUT shopping milk 1
  5. OK
  7. GET shopping milk
  8. 1
  9. OK

但是，由于agent是进程，每个bucket只有一个进程id（pid）而不是名字。
在《入门手册》中进程那章中提到过，我们可以给进程注册名字。
我们可以使用这个方法来给bucket起名：

  1. iex> Agent.start_link(fn -> %{} end, name: :shopping)
  2. {:ok, #PID<0.43.0>}
  3. iex> KV.Bucket.put(:shopping, "milk", 1)
  4. :ok
  5. iex> KV.Bucket.get(:shopping, "milk")
  6. 1

但是这是个很差的办法！在Elixir中，进程的名字存储为原子。
这意味着我们从外部客户端输入的bucket名字，都会被转换成原子。
记住，绝对不要把用户输入转换为原子。这是因为原子不会被垃圾收集器收集。
一旦原子被创建，它就不会被取消（你也没法主动释放一个原子，对吧）。
使用用户输入生成原子就意味着用户可以插入足够不同的名字来耗尽系统内存空间！

在实际操作中，在它用完内存之前会先触及Erland虚拟机的最大原子数量，从而造成系统崩溃。

比起滥用名字注册机制，我们可以创建我们自己的 注册表进程(registry process) 来维护一个字典，
用该字典联系起每个bucket的名字和进程。

这个注册表要能够保证永远处于最新状态。如果有一个bucket进程因故崩溃，注册表必须清除该进程信息，
以防止继续服务下次查找请求。
在Elixir中，我们描述这种情况会说“该注册表需要监视（monitor）每个bucket”。

我们将使用GenServer
来创建一个可以监视bucket进程的注册表进程。

在Elixir和OTP中，GenServer是创建“通用的服务器（generic servers）”的首选抽象物。

3.1-第一个GenServer

一个GenServer实现分为两个部分：客户端API和服务端回调函数。
这两部分可以写在同一个模块里，也可以分开写到两个模块中。
客户端和服务端运行于不同进程，依靠调用客户端函数来与服务端来回传递消息。
方便起见，这里我们将这两部分写在一个模块中。

创建文件lib/kv/registry.ex，包含以下内容：

  1. defmodule KV.Registry do
  2.   use GenServer
  4.   ## Client API
  6.   @doc """
  7.   Starts the registry.
  8.   """
  9.   def start_link() do
  10.     GenServer.start_link(__MODULE__, :ok, [])
  11.   end
  13.   @doc """
  14.   Looks up the bucket pid for `name` stored in `server`.
  16.   Returns `{:ok, pid}` if the bucket exists, `:error` otherwise.
  17.   """
  18.   def lookup(server, name) do
  19.     GenServer.call(server, {:lookup, name})
  20.   end
  22.   @doc """
  23.   Ensures there is a bucket associated to the given `name` in `server`.
  24.   """
  25.   def create(server, name) do
  26.     GenServer.cast(server, {:create, name})
  27.   end
  29.   ## Server Callbacks
  31.   def init(:ok) do
  32.     {:ok, %{}}
  33.   end
  35.   def handle_call({:lookup, name}, _from, names) do
  36.     {:reply, Map.fetch(names, name), names}
  37.   end
  39.   def handle_cast({:create, name}, names) do
  40.     if Map.has_key?(names, name) do
  41.       {:noreply, names}
  42.     else
  43.       {:ok, bucket} = KV.Bucket.start_link()
  44.       {:noreply, Map.put(names, name, bucket)}
  45.     end
  46.   end
  47. end

第一个函数是start_link/0，它传递三个参数启动了一个新的GenServer：

  1. 实现了服务器回调函数的模块名称。这里的__MODULE__指的是当前模块
  2. 初始参数，这里是:ok
  3. 一组选项列表，比如可以存放服务器的名字。这里用个空列表

你可以向一个GenServer发送两种请求：call和cast。Call 是同步的，
服务器 必须 发送回复给该类请求。Cast 是异步的，服务器 不会 发送回复消息。

再往下的两个方法，lookup/2和create/2，它们用来发送这些请求给服务器。
这两种请求，会被第一个参数所指认的服务器中的handle_call/3和handle_cast/2
函数处理（因此你的服务器回调函数必须包含这两个函数）。GenServer.call/2
和GenServer.cast/2除了指认服务器之外，还告诉服务器它们要发送的请求。
这个请求存储在元组里，这里即{:lookup, name}和{:create, name}，
在下面写相应的回调处理函数时会用到。
这个消息元组第一个元素一般是要服务器做的事儿，后面的元素就是该动作的参数。

在服务器这边，我们要实现一系列服务器回调函数来实现服务器的启动、停止以及处理请求等。
回调函数是可选的，我们在这里只实现所关心的那几个。

第一个是init/1回调函数，它接受一个状态参数（你在用户API中调用GenServer.start_link/3中使用的那个），
返回{:ok, state}。这里state是一个新建的图map。
我们现在已经可以观察到，GenServer的API中，客户端和服务器之间的界限十分明显。start_link/3在客户端发生。
而其对应的init/1在服务器端运行。

对于call请求，我们在服务器端必须实现handle_call/3回调函数。
参数：接收某请求（那个元组）、请求来源(_from)以及当前服务器状态（names）。handle_call/3函数返回一个{:reply, reply, new_state}元组。
其中，reply是你要回复给客户端的东西，而new_statue是新的服务器状态。

对于cast请求，我们必须实现一个handle_cast/2回调函数，
接受参数：request以及当前服务器状态（names）。
这个函数返回{:noreply, new_state}形式的元组。

这两个回调函数，handle_call/3和handle_cast/2还可以返回其它几种形式的元组。
还有另外几种回调函数，如terminate/2和code_change/3等。
可以参考完整的GenServer文档来学习相关知识。

现在，来写几个测试来保证我们这个GenServer可以执行预期工作。

3.2-测试一个GenServer

测试一个GenServer比起测试agent没有多少区别。我们在测试的setup回调中启动该服务器进程用以测试。
用以下内容创建测试文件test/kv/registry_test.exs：

  1. defmodule KV.RegistryTest do
  2.   use ExUnit.Case, async: true
  4.   setup do
  5.     {:ok, registry} = KV.Registry.start_link
  6.     {:ok, registry: registry}
  7.   end
  9.   test "spawns buckets", %{registry: registry} do
  10.     assert KV.Registry.lookup(registry, "shopping") == :error
  12.     KV.Registry.create(registry, "shopping")
  13.     assert {:ok, bucket} = KV.Registry.lookup(registry, "shopping")
  15.     KV.Bucket.put(bucket, "milk", 1)
  16.     assert KV.Bucket.get(bucket, "milk") == 1
  17.   end
  18. end

哈，居然都过了！

我们不用显式关闭注册表进程，因为在测试执行完的时候它会自动收到:shutdown信号。
这个方法对于测试是还好啦。如果你想在GenServer的处理逻辑里加上关于停止的方法，
我们可以使用GenServer.stop/1函数：

  1. ## Client API
  3. @doc """
  4. Stops the registry.
  5. """
  6. def stop(server) do
  7.   GenServer.stop(server)
  8. end

3.3-监控需求

至此我们的注册表完成的差不多了，剩下的问题就要解决在有bucket崩溃的时候注册表失去时效的问题。
比如给KV.RegistryTest增加一个测试来暴露这个问题：

  1. test "removes buckets on exit", %{registry: registry} do
  2.   KV.Registry.create(registry, "shopping")
  3.   {:ok, bucket} = KV.Registry.lookup(registry, "shopping")
  4.   Agent.stop(bucket)
  5.   assert KV.Registry.lookup(registry, "shopping") == :error
  6. end

这个测试会在最后一个断言处失败。因为当我们停止了bucket进程后，该bucket名字还存在于注册表中。

为了解决这个bug，我们需要注册表能够监视它派生出的每一个bucket进程。
一旦我们创建了监视器，注册表将收到每个bucket退出的通知。
这样它就可以清理bucket映射字典了。

我们先在命令行中玩弄一下监视机制。启动iex -S mix：

  1. iex> {:ok, pid} = KV.Bucket.start_link
  2. {:ok, #PID<0.66.0>}
  3. iex> Process.monitor(pid)
  4. #Reference<0.0.0.551>
  5. iex> Agent.stop(pid)
  6. :ok
  7. iex> flush()
  8. {:DOWN, #Reference<0.0.0.551>, :process, #PID<0.66.0>, :normal}

注意Process.monitor(pid)返回一个唯一的引用，使我们可以通过这个引用找到其指代的监视器发来的消息。
在我们停止agent之后，我们可以用flush()函数刷新所有消息，
此时会收到一个:DOWN消息，内含一个监视器返回的引用。它表示有个bucket进程退出，
原因是:normal。

现在让我们重新实现下服务器回调函数。

首先，将GenServer的状态改成两个字典：一个用来存储name->pid映射关系，另一个存储ref->name关系。
然后在handle_cast/2中加入监视器，并且实现一个handle_info/2回调函数用来保存监视消息。
下面是修改后完整的服务器调用函数：

  1. ## Server callbacks
  3. def init(:ok) do
  4.   names = %{}
  5.   refs  = %{}
  6.   {:ok, {names, refs}}
  7. end
  9. def handle_call({:lookup, name}, _from, {names, _} = state) do
  10.   {:reply, Map.fetch(names, name), state}
  11. end
  13. def handle_cast({:create, name}, {names, refs}) do
  14.   if Map.has_key?(names, name) do
  15.     {:noreply, {names, refs}}
  16.   else
  17.     {:ok, pid} = KV.Bucket.start_link()
  18.     ref = Process.monitor(pid)
  19.     refs = Map.put(refs, ref, name)
  20.     names = Map.put(names, name, pid)
  21.     {:noreply, {names, refs}}
  22.   end
  23. end
  25. def handle_info({:DOWN, ref, :process, _pid, _reason}, {names, refs}) do
  26.   {name, refs} = Map.pop(refs, ref)
  27.   names = Map.delete(names, name)
  28.   {:noreply, {names, refs}}
  29. end
  31. def handle_info(_msg, state) do
  32.   {:noreply, state}
  33. end

看得出来，我们在没有修改客户端API情况下修改了服务器的实现。
这就体现出了GenServer将客户端与服务器隔离开的好处。

最后，不同于其他回调函数，我们定义了一个“捕捉所有消息”的handle_info/2的函数子句
（可参考《入门》，其意类似重载的函数的一条实现）。它丢弃那些不知道也用不着的消息。
下面一节来解释下为啥。

3.4-call，cast还是info？

到目前为止，我们已经使用了三个服务器回调函数：handle_call/3，handle_cast/2
和handle_info/2。何时使用哪个，其实很直白：

  1. handle_call/3用来处理同步请求。
    这是默认的处理方式，因为等待服务器回复是十分有用的“压力反转（backpressure，涉及IO优化，请自行搜索）”机制。
  2. handle_cast/2用来处理异步请求，当你无所谓要不要个回复时。
    一个cast请求甚至不保证服务器收到了该请求，因此请有节制地使用。
    例如，我们定义的create/2函数应该使用call的，而我们用cast只是为了演示目的。
  3. handle_info/2用来接收和处理服务器收到的其它
    （既不是GenServer.call/3也不是GenServer.cast/2）请求。
    它可以接受是以普通进程身份通过send/2收到的消息或者其它消息。监视器发来的:DOWN消息就是个极好的例子。

因为任何消息，包括通过send/2发送的消息，回去到handle_info/2处理，
因此便会有很多你不需要的消息跑进服务器。如果不定义一个“捕捉所有消息”的函数子句，
这些消息会导致我们的监督者进程（supervisor）崩溃，因为没有函数子句匹配它们。

我们不需要为handle_call/3和handle_cast/2担心这个情况，
因为它们能接受的请求都是通过GenServer的API发送的，要是出了毛病就是程序员自己犯错。

3.5-监视器还是链接？

我们之前在 进程 那章里的学习过链接（links）。现在，随着注册表的完工，
你也许会问：我们啥时候用监控器，啥时候用链接呢？

链接是双向的。你将两个进程链接起来，其中一个挂了，另一个也会挂（除非它处理了该异常，改变了行为）。
而监视机制是单向的：只有监视别人的进程会收到被监视的进程的消息。
简单说，当你想让某些进程一挂都挂时，使用链接；而想要得到进程退出或挂了等事件的消息通知，使用监视。

回到我们handle_cast/2的实现，你可以看到注册表是同时链接着且监视着派生出的bucket：

  1. {:ok, pid} = KV.Bucket.start_link()
  2. ref = Process.monitor(pid)

这是个坏主意。我们不想注册表进程因为某个bucket进程挂而一同挂掉！
我们将在讲解监督者（supervisor）时探索更好的解决方法。
一句话概括，我们将不直接创建新的进程，而是将把这个责任委托给监督者。
就像我们即将看到的那样，监督者同链接工作在一起，这就解释了为啥基于链接的API
（如spawn_link，start_link等）在Elixir和OTP上十分流行。

在讲监督者之前，我们首先探索下使用GenEvent进行事件管理以和处理的知识。

免责声明： 以上内容版权归原作者所有，如有侵犯您的原创版权请告知，我们将尽快删除相关内容。感谢每一位辛勤著写的作者，感谢每一位的分享。
@小牛知识库 xnip.cn