EMZETT.
Login

Emzett Wiki

Elixir: Prozesse, GenServer und OTP

Properties1
tagsprogrammiersprachen, elixir

4 Min. Lesezeit

Kurz: Elixir-Prozesse sind keine Betriebssystem-Prozesse, sondern winzige Einheiten der BEAM (wenige KB), von denen man Hunderttausende starten kann. Sie teilen keinen Speicher und kommunizieren nur über Nachrichten.

Teil des Kurses Elixir

Kapitel 6 von 8 im Kurs Elixir (Abschnitt „Nebenläufigkeit“). Mit Fortschritt, Quiz und Zertifikat auf der Lernseite.

Prozesse

Elixir-Prozesse sind keine Betriebssystem-Prozesse, sondern winzige Einheiten der BEAM (wenige KB), von denen man Hunderttausende starten kann. Sie teilen keinen Speicher und kommunizieren nur über Nachrichten.

pid = spawn(fn -> IO.puts("Hallo aus einem Prozess") end)
Process.sleep(50)
IO.inspect(is_pid(pid))
IO.inspect(Process.alive?(pid))
 
eltern = self()
spawn(fn -> send(eltern, {:ergebnis, 6 * 7}) end)
receive do
  {:ergebnis, wert} -> IO.puts("Antwort: #{wert}")
after
  1000 -> IO.puts("Timeout")
end
 
# Ein Prozess, der Nachrichten verarbeitet
defmodule Echo do
  def schleife do
    receive do
      {:echo, absender, text} ->
        send(absender, {:antwort, String.upcase(text)})
        schleife()
      :stop ->
        :ok
    end
  end
end
 
echo = spawn(fn -> Echo.schleife() end)
send(echo, {:echo, self(), "hallo"})
receive do
  {:antwort, text} -> IO.puts(text)
end
send(echo, :stop)

Ausgabe:

Hallo aus einem Prozess
true
false
Antwort: 42
HALLO

Viele Prozesse parallel: Task

Task ist die komfortable Abkürzung für “etwas nebenläufig berechnen und das Ergebnis abholen”:

aufgaben = for n <- 1..5, do: Task.async(fn -> Process.sleep(20); n * n end)
IO.inspect(Task.await_many(aufgaben))
 
ergebnis =
  1..20
  |> Task.async_stream(fn n -> n * 2 end, max_concurrency: 4)
  |> Enum.map(fn {:ok, wert} -> wert end)
  |> Enum.sum()
IO.inspect(ergebnis)
 
# 10.000 Prozesse starten
prozesse = for i <- 1..10_000, do: spawn(fn -> receive do :ende -> i end end)
IO.inspect(length(prozesse))
IO.inspect(Process.info(self(), :message_queue_len))
Enum.each(prozesse, &send(&1, :ende))

Ausgabe:

[1, 4, 9, 16, 25]
420
10000
{:message_queue_len, 0}

Agent: einfacher Zustand

{:ok, zaehler} = Agent.start_link(fn -> 0 end)
Agent.update(zaehler, &(&1 + 1))
Agent.update(zaehler, &(&1 + 10))
IO.inspect(Agent.get(zaehler, & &1))
Agent.get_and_update(zaehler, fn s -> {s, s * 2} end) |> IO.inspect()
IO.inspect(Agent.get(zaehler, & &1))
 
tasks = for _ <- 1..100, do: Task.async(fn -> Agent.update(zaehler, &(&1 + 1)) end)
Task.await_many(tasks)
IO.inspect(Agent.get(zaehler, & &1))

Ausgabe:

11
11
22
122

GenServer: Server-Prozess mit Zustand

Der GenServer (aus OTP, Open Telecom Platform) ist der Standardbaustein für Prozesse mit Zustand, ein Client schickt call (mit Antwort) oder cast (ohne Antwort):

defmodule Warenkorb do
  use GenServer
 
  # Client-API
  def start_link(opts \\ []), do: GenServer.start_link(__MODULE__, %{}, opts)
  def hinzufuegen(pid, artikel, menge \\ 1), do: GenServer.cast(pid, {:hinzufuegen, artikel, menge})
  def inhalt(pid), do: GenServer.call(pid, :inhalt)
  def anzahl(pid), do: GenServer.call(pid, :anzahl)
  def leeren(pid), do: GenServer.call(pid, :leeren)
 
  # Server-Callbacks
  @impl true
  def init(zustand), do: {:ok, zustand}
 
  @impl true
  def handle_cast({:hinzufuegen, artikel, menge}, zustand) do
    {:noreply, Map.update(zustand, artikel, menge, &(&1 + menge))}
  end
 
  @impl true
  def handle_call(:inhalt, _von, zustand), do: {:reply, zustand, zustand}
  def handle_call(:anzahl, _von, zustand), do: {:reply, zustand |> Map.values() |> Enum.sum(), zustand}
  def handle_call(:leeren, _von, zustand), do: {:reply, zustand, %{}}
end
 
{:ok, korb} = Warenkorb.start_link()
Warenkorb.hinzufuegen(korb, "Apfel", 3)
Warenkorb.hinzufuegen(korb, "Birne")
Warenkorb.hinzufuegen(korb, "Apfel", 2)
IO.inspect(Warenkorb.inhalt(korb))
IO.inspect(Warenkorb.anzahl(korb))
IO.inspect(Warenkorb.leeren(korb))
IO.inspect(Warenkorb.inhalt(korb))

Ausgabe:

%{"Apfel" => 5, "Birne" => 1}
6
%{"Apfel" => 5, "Birne" => 1}
%{}

Supervisor: “Let it crash”

Statt Fehler überall abzufangen, lässt man Prozesse abstürzen und Supervisoren sie neu starten. Das macht Systeme robust:

defmodule Fragil do
  use GenServer
  def start_link(_), do: GenServer.start_link(__MODULE__, 0, name: __MODULE__)
  def abstuerzen, do: GenServer.cast(__MODULE__, :boom)
  def zaehlen, do: GenServer.call(__MODULE__, :zaehlen)
  @impl true
  def init(n), do: {:ok, n}
  @impl true
  def handle_cast(:boom, _), do: raise("Absturz!")
  @impl true
  def handle_call(:zaehlen, _, n), do: {:reply, n + 1, n + 1}
end
 
Logger.configure(level: :emergency)    # Absturz-Berichte hier ausblenden
{:ok, sup} = Supervisor.start_link([Fragil], strategy: :one_for_one)
IO.inspect(Fragil.zaehlen())
IO.inspect(Fragil.zaehlen())
alt = Process.whereis(Fragil)
Process.flag(:trap_exit, true)
Fragil.abstuerzen()
Process.sleep(100)
neu = Process.whereis(Fragil)
IO.inspect(alt != neu)
IO.inspect(Fragil.zaehlen())
IO.inspect(Supervisor.count_children(sup).active)

Ausgabe:

1
2
true
1
1
StrategieVerhalten
:one_for_onenur der abgestürzte Prozess wird neu gestartet
:one_for_allalle Kinder werden neu gestartet
:rest_for_oneder abgestürzte und alle danach gestarteten

Merke

  • Prozesse sind leichtgewichtig, isoliert und kommunizieren per Nachrichten (send/receive)
  • Task für nebenläufige Berechnungen, Agent für einfachen Zustand
  • GenServer für Server mit Zustand: call (synchron), cast (asynchron)
  • Supervisoren starten abgestürzte Prozesse neu: “let it crash”

Übungsaufgabe

Baue einen GenServer, der einen Zähler hält und inkrement und wert anbietet.

Quiz zur Selbstkontrolle

Weiter im Kurs

Zurück: Structs, Protokolle, Behaviours und Fehler

Weiter: Mix, Tests und Phoenix

Alle Kapitel: Elixir im Überblick

Graphansicht

Backlinks