ElixirSwarm浅析

  • Z时代
  • 2024-01-10
  • 分类：综合

编程

Elixir是一门很有意思的语言，它的社区把很多的关注重点放在了解决集群化带来的挑战上。libcluster和swarm就是为此诞生的两个库。但是由于这门语言还很年轻，很多文档还很不成熟（Swarm的文档就是这样），所以有很多时候你要自己亲手做实验才能明白其中的门道。今天我们就来聊聊Swarm这个库。

前言

也许你早就听说过CAP定律。简单说，就是“一致性（Consistency）、可用性（Availability）、分区容错性（Partition tolerance）三者在一个分布式数据容器里最多只能要两个”。

Elixir Swarm给了我们选择CP（一致性+分区容错性）或AP（可用性+分区容错性）的自由。

Elixir Swarm基础

我们就做一个分布式计数器，它只支持两个操作：获取当前值（value）和+1（increment）。首先创建一个项目：

$ mix new --sup counter

然后创建一个worker模块。我这里用GenServer而不用Agent是因为接下来会往里加Elixir Swarm生命周期的钩子，如果用Agent不方便。

defmodule Counter.Worker do
  use GenServer
  def start_link(_) do
    GenServer.start_link(__MODULE__, 0)
  end
  def value do
    GenServer.call({:via, :swarm, __MODULE__}, :value)
  end
  def increment do
    GenServer.cast({:via, :swarm, __MODULE__}, :increment)
  end
  @impl true
  def init(initial_state) do
    {:ok, initial_state}
  end
  @impl true
  def handle_call(:value, _from, state) do
    {:reply, state, state}
  end
  @impl true
  def handle_cast(:increment, state) do
    {:noreply, state + 1}
  end
end

相信写个计数器对大家都不是难事。这里要注意的是，GenServer.cast和GenServer.call的第一个参数是{:via, :swarm, 进程在Elixir Swarm里注册的名称}。也许你会问“为什么不用GenServer.start_link里指定的名称”，那是因为如果用那个名称，则Erlang运行时会在当前节点找进程，而在Elixir Swarm里，这个进程很可能不在当前节点！我们要委托Elixir Swarm帮我们找进程，所以当然要用它在Elixir Swarm里注册的名称啦。

由于Elixir Swarm管理的进程并不会自动被加进supervision tree，所以我们要创建一个supervisor并显式地把它加进supervision tree。另外，Elixir Swarm需要你手动注册进程创建手段，然后它接管进程的所有生命周期，所以我们的supervisor不需要预先有子进程，也就是说，我们需要一个DynamicSupervisor。

defmodule Counter.Supervisor do
  use DynamicSupervisor
  def start_link(_) do
    DynamicSupervisor.start_link(__MODULE__, nil, name: __MODULE__)
  end
  def init(_) do
    DynamicSupervisor.init(strategy: :one_for_one)
  end
  @doc """
  这个函数会作为`Counter.Worker`进程的创建函数被注册进Elixir Swarm。
  此函数必须在成功创建进程时返回`{:ok, pid}`。
  返回的pid会被Elixir Swarm管理生命周期。
  """
  def register(child_spec) do
    DynamicSupervisor.start_child(__MODULE__, child_spec)
  end
end

这只是一个最最普通的dynamic supervisor。这个模块会在每个节点创建一个supervisor进程。

接下来处理注册。在lib/counter/application.ex里，

defmodule Counter.Application do
  def start(_type, _args) do
    children = [
      Counter.Supervisor
    ]
    {:ok, sup} = Supervisor.start_link(children, strategy: :one_for_one, name: Couter.TopSupervisor)
    Swarm.register_name(Counter.Worker, Counter.Supervisor, :register, [Counter.Worker])
    {:ok, sup}
  end
end

Swarm.register_name的参数：

  1. 进程注册的名称

  2. 负责创建进程的模块

  3. 上述模块中负责创建进程的函数

  4. 调用上述函数时传入的参数列表

  5. 超时（单位：毫秒，可选，默认:infinity）

注意，Swarm.register_name必须在Counter.Supervisor正常启动后执行（好象是废话）。另外，我们不应该模式匹配Swarm.register_name的返回值，理由是在一个分布式系统里，各个节点的启动会有时间差。如果一个进程在某个节点注册成功，在其他节点就会注册失败，因为同名进程已经注册过了，但这并不表示其他节点不应该正常启动。

由此也可以看出，正常情况下，在一个集群里，不会有两个同名进程。这里所说的“同名”的“名”指的是在Elixir Swarm里注册的名称，即Swarm.register_name的第一个参数，而不是GenServer.start_link里的:name选项。注意我说的“正常情况”，因为在“不正常情况”下在集群里会有多个同名进程哦。

网络断裂（netsplit）和进程移交（handoff of processes）

注意：Elixir Swarm官方README的示例是错误的！！

网络断裂又称网络分区（network partition），是指集群中的部分节点和其他节点失联但仍然持续运行的异常情况。

Elixir Swarm会保证在每个注册的名称在每个分区都有一个进程，所以在这种情况下，会出现上述的多个同名进程的情况。

当出现网络断裂时，Elixir Swarm会监测到“网络拓扑变更”，具体机制和配置有关（后述），并作出相应的处理。

  • 在注册进程不存在的分区，根据注册名称的哈希值计算进程应在的节点，并根据注册时的信息在该节点创建新进程。

  • 在注册进程已存在的分区，重新计算进程应在的节点，如果该节点不是进程所在的当前节点，则在目标节点上创建新进程，并让老进程把当前状态移交给新进程，然后杀死老进程。

从这里我们可以看出两点：

  • Elixir Swarm并不提供进程的副本，也不提供从副本恢复数据的机制！

  • 各个网络分区中的同名进程的状态是不一样的！

如果你不能接受这个现实，则你应该考虑使用其他库，例如Raft算法的Erlang实现：rabbitmq/ra。

处理进程移交的代码如下：

defmodule Counter.Worker do
  [...]
  @doc """
  这个回调会在老进程死之前被Elixir Swarm调用，
  用来获取老进程的需要移交的状态。
  这里我把整个状态都移交出去了。
  """
  def handle_call({:swarm, :begin_handoff}, _from, local_state) do
    {:reply, {:resume, local_state}, local_state}
  end
  @doc """
  这个回调会在新进程里被调用，
  用来接收并合并老进程移交过来的状态。
  这里我只是简单地抛弃本地状态，完全采用老进程状态。
  """
  def handle_cast({:swarm, :end_handoff, remote_state}, _local_state) do
    {:noreply, remote_state}
  end
  @doc """
  这个回调会在老进程死之前被调用。
  你可以在这里做一些善后工作（比如清除本地硬盘上的状态文件之类的）。
  我这里只是简单地让进程优雅地死去。
  """
  def handle_info({:swarm, :die}, state) do
    {:stop, :shutdown, state}
  end
end

当网络断裂愈合后，Elixir Swarm会尝试解决冲突并在多个同名进程中保留一个，可能会发生进程移交（未考证）。处理冲突的代码如下：

defmodule Counter.Worker do
  [...]
  @doc """
  这个回调会在网络断裂愈合时被调用。
  我这里只是简单地取两个计数中较大的作为冲突解决后的状态。
  """
  def handle_cast({:swarm, :resolve_conflict, remote_state}, local_state) do
    {:noreply, if(remote_state > local_state, do: remote_state, else: local_state)}
  end
end

不要吐槽我的实现。要完美处理冲突从来不是件容易的事，你需从数据结构开始重新考虑问题。我这里不展开了。

如果跑进程的那个节点宕机了，会发生什么？

很简单，在某个活着的节点上重新启动进程，仅此而已。

那之前的状态呢？

丢了呗。如果你不能接受这个现实，请考虑持久化或者PAXOS/RAFT。这不是Elixir Swarm能解决的问题。

关于Elixir Swarm的分布策略

Elixir Swarm提供了两种分布策略：Ring和StaticQuorumRing。这两者的区别是什么？

Ring

当客户端进程访问注册的服务器进程时，客户端进程访问到的是自己所在的网络分区中的服务器进程，因此它获取到的数据可能是不对的，但能获取到。如果客户端进程想要修改服务器进程的状态，它改到的也是自己所在的分区里的服务器进程状态。当网络断裂愈合时，很可能因此发生冲突。

如果你选择Ring，则你选择了AP系统。

StaticQuorumRing

当客户端进程访问注册的服务器进程时，如果客户端所在分区的节点数小于所配置的:static_quorum_size，则访问会失败。推荐把:static_quorum_size的值设置成集群节点数量 / 2 + 1。

如果你选择StaticQuorumRing并按上述配置:static_quorum_size，则你选择了CP系统。

以上是 ElixirSwarm浅析 的全部内容， 来源链接： utcz.com/z/514665.html

回到顶部