클러스터에 ETS 데이터베이스 배포

Erlang 생태계의 내장형 빠른 인메모리 데이터베이스인 ETS를 배포하는 방법을 설명하는 간단한 글을 제공합니다.

분산 데이터가 필요한 이유는 무엇입니까? "확장된"웹 앱과 상호 작용할 때 도달할 노드를 알 수 없으므로 모든 노드/포드에 동일한 데이터가 있어야 합니다. 일부 데이터는 클라이언트 측(예: 장바구니)에 보관할 수 있지만 클라이언트 자격 증명이나 캐시와 같은 일부 데이터는 서버 측에도 보관할 수 있습니다.

빠른 읽기 데이터를 관리하는 몇 가지 방법:

Redis 와 같은 외부 데이터베이스를 사용하여 앱을 상태 비저장으로 만듭니다. 그러나 Redis를 보호하고 클러스터를 만들려는 경우 특히 눈에 띄는 오버헤드가 발생합니다.

앱을 상태 저장(즉, 데이터를 로컬에 저장)하고 로드 밸런서 또는 Kubernetes의 경우 수신 친화성을 사용하여 고정 세션을 구현합니다.

앱을 상태 저장 상태로 유지하고 노드를 클러스터링합니다. 로컬 데이터베이스는 Kubernetes를 사용하는지 여부에 관계없이 동기화, 리버스 프록시 여부를 지정해야 합니다.

고정 세션 사용의 단점은 새 사용자만 새 노드에 도달하기 때문에 노드 간에 로드를 분산할 수 없다는 것입니다.

BEAM은 기본적으로 클러스터링(Elixir/Phoenix를 실행하는 Erlang의 가상 머신)을 지원하기 때문에 이 마지막 경로를 사용합니다. 클러스터 모드에서는 분산형PG이 있는 PubSub 모듈이 무료입니다. 이것은 Erlang 생태계의 두 번째 주요 지점입니다. 덕분에 단일 실패 지점이 없고 외부 종속성이 없습니다.

Why would you choose ETS and not MNESIA which is a build-in distributed solution? The main reason was the setup of MNESIA in a dynamic environment such as Kubernetes. Bad synchronisation and proper startup on node discovery can be difficult.

MNESIA에는 메모리 또는 디스크의 두 가지 모드가 있고 ETS에는 디스크 복사 분신인 DETS가 있습니다. 여전히 ETS 테이블의 디스크 복사본을 만들 수 있습니다. 테이블 이름이 :users 이면 :ets.tab2file(:users, 'data.txt') 명령이 디스크에 저장됩니다(! 작은따옴표 사용 !).

사용 사례의 경우 libcluster와 함께 노드 검색을 사용하고 노드 시작 시 PubSub를 통해 ETS를 동기화하는 것이 쉽고 안정적이었습니다. 1,000명의 회원을 쉽게 팔로우할 수 있습니다.

설명할 코드가 있습니다. 사용자를 생성하고 ETS에 저장합니다. 우리는 GenServer 동작을 사용합니다. 우리는 GenServer 메모리에 상태를 유지하지 않고(실제로 상태는 단지 nil) 오히려 GenServer에서 제공하는 프로세스 간의 메시징 기능을 사용합니다.

Since the ETS database is configured as a set, they will be no duplicate or data lost.

앱에서 ETS 작업에 저장하고(예: 사용자가 생성될 때) 키:new를 통해 주제 "new"에 이 데이터를 브로드캐스트하는 작업을 수행하는 "Repo"모듈이 있습니다.

defmodule MyApp.Repo do

  require Logger
  alias :ets, as: Ets

  def all,
    do: Ets.tab2list(:users)

  def save(message),
    do: true = Ets.insert(:users, message)

  def save_and_emit(email, context) do
    user = build_from{email, token)
    true = save(user)
    :ok = Phoenix.PubSub.broadcast_from!(PwdlessGs.PubSub, self(), "new", {:new, user})
    user
  end
end

노드(ErlangEPMD 통합 서버)를 모니터링하는 GenServer "NodeListener"가 있습니다. 그의 임무는 :nodeup 이벤트를 수신하고 :sync 키를 통해 "node_up"주제의 다른 노드에 ETS 테이블을 브로드캐스트하는 것입니다.

defmodule MyApp.NodeListener do
  use GenServer
  alias MyApp.Repo

  # delay between the "up" event and the node read
  @sync_init 3_000

  def start_link(_opts),
    do: GenServer.start_link(__MODULE__,[], name: __MODULE__)

  @impl true
  def init([]) do
    :ok = :net_kernel.monitor_nodes(true)
    {:ok, nil}
  end

  @impl true
  def handle_info({:nodeup,_}, _) do
    Process.send_after(self(), {:perform_sync}, @sync_init)
    {:noreply, nil}
  end

  @impl true
  def handle_info({:perform_sync}, _state) do
    :ok = Phoenix.PubSub.broadcast_from!(MyApp.PubSub, self(), "node_up", {:sync, Repo.all()}, node())
    {:noreply, nil}
  end

  def handle_info({:nodedown, _},_), do: {:noreply, nil}
end

"node_up"과 "new"라는 두 개의 토픽을 구독하는 GenServer "Syncer"가 있습니다. 그에게는 세 가지 임무가 있습니다.

시작 시 자신의 데이터 브로드캐스트(클러스터에 데이터가 있는 노드를 업로드할 수 있음),

은 :sync 처리기를 듣습니다. 그런 다음 노드는 n-1개의 복사본(n개 노드 중)을 수신하고 데이터를 자신의 테이블과 병합합니다.

은 :new 처리기를 듣습니다. 각 노드(예: 새 사용자)의 작업 변경 사항을 전달합니다.

defmodule MyApp.Syncer do
  use GenServer
  require Logger
  alias :ets, as: Ets
  alias MyApp.Repo

  def load_data(loading) do
    if loading do
      {:ok, _} = Task.start(fn -> 
        fetch_data() |> Repo.save()
        :ok = Phoenix.PubSub.broadcast_from!(
          MyApp.PubSub,
          self(),
          "node_up",
          {:sync, Repo.all(), node()}
        )
      end)
    end

    :ok
  end

  def start_link(opts) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end

  @impl true
  def init(opts) do
    :users = Ets.new(:users, [:set, :public, :named_table, keypos: 1])
    Logger.info("ETS table started...")

    :ok = Phoenix.PubSub.subscribe(MyApp.PubSub, "new")
    :ok = Phoenix.PubSub.subscribe(MyApp.PubSub, "node_up")

    :ok = if (opts[:preload]),
      do: load_data(true)

    {:ok, nil}
  end

  @impl true
  def handle_info({:new, user}, _state) do
    Repo.save(user)
    {:noreply, nil}
  end

  @impl true
  def handle_info({:sync, table, node, message}, _state) do
    if node != node(), do: Repo.save(table)

    {:noreply, nil}
  end
end

구성에 대한 설명

(이 순서대로) 다음과 같이 감독됩니다.

{Phoenix.PubSub, name: MyApp.PubSub, adapter: Phoenix.PubSub.PG2},
MyApp.NodeListener,
{MyApp.Syncer, [preload: true]},

구성은 다음과 같습니다.

# config.exs
config :my_app, MyAppWeb.Endpoint,
  [...],
  pubsub_server: MyApp.PubSub,