任务与gen_tcp

  1. 回显服务器
  2. 任务
  3. 任务主管

本章,我们将学习如何使用Erlang的:gen_tcp模块来处理请求.这是一个探索Elixir的Task模块的好机会.在后面的章节我们将扩展我们的服务器,让它能够执行命令.

回显服务器(Echo server)

我们将实现一个回显服务器,作为我们的TCP服务器的开始.它会简单地发送一个回复,内容就是在请求中收到的文本.我们将缓慢地升级服务器,直到它被监督并准备好处理多重连接.

一个TCP服务器,概括地说,会有如下表现:

  1. 监听一个端口,直到端口可用,并获得套接字
  2. 在那个端口等待客户端连接并接受它
  3. 读取客户端请求并进行回复

让我们来实现这些步骤.进入apps/kv_server应用,打开lib/kv_server.ex,并添加下列函数:

  1. require Logger
  3. def accept(port) do
  4.   # 下列选项的意思是:
  5.   #
  6.   # 1. `:binary` - 以二进制数接受数据 (而非列表)
  7.   # 2. `packet: :line` - 逐行接收数据
  8.   # 3. `active: false` - 阻塞在 `:gen_tcp.recv/2` 直到数据可用
  9.   # 4. `reuseaddr: true` - 允许我们重用数据当监听器崩溃时
  10.   {:ok, socket} = :gen_tcp.listen(port,
  11.                     [:binary, packet: :line, active: false, reuseaddr: true])
  12.   Logger.info "Accepting connections on port #{port}"
  13.   loop_acceptor(socket)
  14. end
  16. defp loop_acceptor(socket) do
  17.   {:ok, client} = :gen_tcp.accept(socket)
  18.   serve(client)
  19.   loop_acceptor(socket)
  20. end
  22. defp serve(socket) do
  23.   socket
  24.   |> read_line()
  25.   |> write_line(socket)
  27.   serve(socket)
  28. end
  30. defp read_line(socket) do
  31.   {:ok, data} = :gen_tcp.recv(socket, 0)
  32.   data
  33. end
  35. defp write_line(line, socket) do
  36.   :gen_tcp.send(socket, line)
  37. end

我们将通过调用KVServer.accept(4040)来开启我们的服务器,这里的4040是端口.accept/1的第一步就是监听端口,直到套接字可用,然后调用loop_acceptor/1.loop_acceptor/1只是一个接收客户端连接的循环.对于每个被接收的连接,我们调用serve/1.

serve/1是另一个循环,它从套接字中读取一行并将那些行写回套接字中.注意serve/1函数使用了管道操作符|>来表示这个操作流.管道操作符的左边会被执行,并将结果作为右边函数的第一个参数.上面的例子:

  1. socket |> read_line() |> write_line(socket)

等同于:

  1. write_line(read_line(socket), socket)

read_line/1通过:gen_tcp.recv/2实现了从套接字中接收数据,而write_line/2使用:gen_tcp.send/2向套接字中写入.

这些就是我们实现回显服务器所需的一切.让我们来试一试!

通过iex -S mix在kv_server应用中运行一个IEx会话.在IEx中,运行:

  1. iex> KVServer.accept(4040)

现在服务器已经运行了,你会发现控制台阻塞了.让我们使用一个telnet客户端来访问我们的服务器.这些客户端在大多数操作系统中都是可用的,它们的命令行也是类似的:

  1. $ telnet 127.0.0.1 4040
  2. Trying 127.0.0.1...
  3. Connected to localhost.
  4. Escape character is '^]'.
  5. hello
  6. hello
  7. is it me
  8. is it me
  9. you are looking for?
  10. you are looking for?

输入”hello”,敲击回车,将返回”hello”.完美!

我的telnet客户端可以通过输入ctrl + ],输入quit,再敲击<Enter>来退出,但你的客户端可能步骤会有不同.

一旦你退出了telnet客户端,你会在IEx会话中看到一个错误:

  1. ** (MatchError) no match of right hand side value: {:error, :closed}
  2.     (kv_server) lib/kv_server.ex:41: KVServer.read_line/1
  3.     (kv_server) lib/kv_server.ex:33: KVServer.serve/1
  4.     (kv_server) lib/kv_server.ex:27: KVServer.loop_acceptor/1

这是因为我们期望从:gen_tcp.recv/2中获得数据,但客户端关闭了连接.我们将在之后对服务器的修改中更好地处理这种情形.

现在,这里有一个更加重要的bug需要我们去修复:当我们的TCP接收器崩溃时会发生什么?因为这里没有监督,所以服务器死亡后,我们不能再处理更多请求,因为它不会重启.这就是为什么我们必须将服务器移动到一个监督树上.

任务(Tasks)

我们已经学习了代理,通用服务器,以及主管.它们都需要处理多重的消息或消息状态.但是当我们只需要执行一些任务时应该使用什么呢?

Task模块提供了这个功能.例如,它有一个start_link/3函数用于接收模块,函数和参数,允许我们运行一个给定的函数作为监督树的一部分.

来试一下.打开lib/kv_server.ex,将start/2函数中的主管修改成:

  1. def start(_type, _args) do
  2.   import Supervisor.Spec
  4.   children = [
  5.     worker(Task, [KVServer, :accept, [4040]])
  6.   ]
  8.   opts = [strategy: :one_for_one, name: KVServer.Supervisor]
  9.   Supervisor.start_link(children, opts)
  10. end

通过这个改动,我们将KVServer.accept(4040)作为一个工人来运行.我们一直在编写端口,而等一下我们将会讨论它可以被如何改动.

现在服务器是监督树的一部分了,它会在我们运行应用时自动启动.在终端中输入mix run --no-halt,并再次使用telnet客户端来确认一切正常:

  1. $ telnet 127.0.0.1 4040
  2. Trying 127.0.0.1...
  3. Connected to localhost.
  4. Escape character is '^]'.
  5. say you
  6. say you
  7. say me
  8. say me

正常!如果你杀死客户端,导致整个服务器崩溃,你会看到另一个立刻启动了.然而,它是规模化(scale)的吗?

试着同时连接两个telnet客户端.当你这么做时,你会发现,第二个客户端没有回声:

  1. $ telnet 127.0.0.1 4040
  2. Trying 127.0.0.1...
  3. Connected to localhost.
  4. Escape character is '^]'.
  5. hello
  6. hello?
  7. HELLOOOOOO?

它似乎完全不能工作.原因是我们接受连接和处理请求是在同一个进程中.当一个客户端连接上,我们就不能再接受其它客户端了.

任务主管(Task Supervisor)

为了使我们的服务器能够处理同时连接,我们需要有一个接收器进程,它能够生成其它进程来处理请求.一个方法要修改一下:

  1. defp loop_acceptor(socket) do
  2.   {:ok, client} = :gen_tcp.accept(socket)
  3.   serve(client)
  4.   loop_acceptor(socket)
  5. end

要使用Task.start_link/1,它类似于Task.start_link/3,但是它接收的是一个匿名函数而非模块,函数和参数:

  1. defp loop_acceptor(socket) do
  2.   {:ok, client} = :gen_tcp.accept(socket)
  3.   Task.start_link(fn -> serve(client) end)
  4.   loop_acceptor(socket)
  5. end

我们直接在接收器进程中启动了一个链接任务.但我们已经犯过这种错误了.你还记得吗?

这类似于我们从注册表中直接调用KV.Bucket.start_link/0时所犯的错误.那意味着任何桶的失败都会导致整个注册表崩溃.

上述代码也有同样的缺陷:如果我们将serve(client)任务链接到接收器,一个处理请求时的崩溃将会传递给接收器,从而导致其它所有连接失败.

我们为注册表修正了这个问题,通过使用一个简单地一对一主管.我们在此将如法炮制,由于这个模式太常用于任务了,Task已经提供了一个方案:在我们的监督树上,可以使用一个简单的一对一主管,附带临时的工人.

让我们再次修改start/2,添加一个主管到我们的树上:

  1. def start(_type, _args) do
  2.   import Supervisor.Spec
  4.   children = [
  5.     supervisor(Task.Supervisor, [[name: KVServer.TaskSupervisor]]),
  6.     worker(Task, [KVServer, :accept, [4040]])
  7.   ]
  9.   opts = [strategy: :one_for_one, name: KVServer.Supervisor]
  10.   Supervisor.start_link(children, opts)
  11. end

我们简单地启动了一个名为KVSever.TaskSupervisor的Task.Supervisor进程.记住,因为接受其任务依赖于该主管,所以主管必须先启动.

现在我们只需要修改loop_acceptor/1来使用Task.Supervisor处理每个请求:

  1. defp loop_acceptor(socket) do
  2.   {:ok, client} = :gen_tcp.accept(socket)
  3.   {:ok, pid} = Task.Supervisor.start_child(KVServer.TaskSupervisor, fn -> serve(client) end)
  4.   :ok = :gen_tcp.controlling_process(client, pid)
  5.   loop_acceptor(socket)
  6. end

你可能注意到我们添加了一行,:ok = :gen_tcp.controlling_process(client, pid).这使得子进程成为了client套接字的”控制进程”.如果不这样做的话,一旦接收器崩溃,它就会关闭所有客户端,因为套接字默认被绑定到了那个接收它们的进程上.

通过mix run --no-halt启动一个新的服务器,现在我们可以打开许多并发的telnet客户端了.你也会发现退出一个客户端不会导致接收器崩溃了.优秀!

这里是完整的回显服务器实现,在单独的模块中:

  1. defmodule KVServer do
  2.   use Application
  3.   require Logger
  5.   @doc false
  6.   def start(_type, _args) do
  7.     import Supervisor.Spec
  9.     children = [
  10.       supervisor(Task.Supervisor, [[name: KVServer.TaskSupervisor]]),
  11.       worker(Task, [KVServer, :accept, [4040]])
  12.     ]
  14.     opts = [strategy: :one_for_one, name: KVServer.Supervisor]
  15.     Supervisor.start_link(children, opts)
  16.   end
  18.   @doc """
  19.   Starts accepting connections on the given `port`.
  20.   """
  21.   def accept(port) do
  22.     {:ok, socket} = :gen_tcp.listen(port,
  23.                       [:binary, packet: :line, active: false, reuseaddr: true])
  24.     Logger.info "Accepting connections on port #{port}"
  25.     loop_acceptor(socket)
  26.   end
  28.   defp loop_acceptor(socket) do
  29.     {:ok, client} = :gen_tcp.accept(socket)
  30.     {:ok, pid} = Task.Supervisor.start_child(KVServer.TaskSupervisor, fn -> serve(client) end)
  31.     :ok = :gen_tcp.controlling_process(client, pid)
  32.     loop_acceptor(socket)
  33.   end
  35.   defp serve(socket) do
  36.     socket
  37.     |> read_line()
  38.     |> write_line(socket)
  40.     serve(socket)
  41.   end
  43.   defp read_line(socket) do
  44.     {:ok, data} = :gen_tcp.recv(socket, 0)
  45.     data
  46.   end
  48.   defp write_line(line, socket) do
  49.     :gen_tcp.send(socket, line)
  50.   end
  51. end

由于我们改变了主管的规则,我们不禁要问:我们的监督策略还正确吗?

在这种情况下,是正确的:如果接收器崩溃,不需要关闭现存的连接.换句话说,如果任务主管崩溃,也不需要关闭接收器.

下一章我们将开始解析客户端请求和发送回复,完成我们的服务器.