Elixir进程通信 #

iex(1)> send(self(), :hello)
:hello

iex(1)> receive do
...>   :hello -> "Got hello!"
...>   :world -> "Got world!"
...> end
"Got hello!"

iex(1)> receive do
...>   msg -> msg
...> after
...>   1000 -> {:error, :timeout}
...> end
{:error, :timeout}

iex(1)> receive do
...>   {:ok, data} -> "Success: #{data}"
...>   {:error, reason} -> "Error: #{reason}"
...>   {from, msg} when is_pid(from) -> "From #{inspect(from)}: #{msg}"
...> end

┌─────────────────┐
│   Message 1     │
├─────────────────┤
│   Message 2     │
├─────────────────┤
│   Message 3     │
├─────────────────┤
│      ...        │
└─────────────────┘

iex(1)> send(self(), :msg1)
iex(2)> send(self(), :msg2)
iex(3)> send(self(), :msg3)

iex(4)> Process.info(self(), :messages)
{:messages, [:msg1, :msg2, :msg3]}

def flush do
  receive do
    _ -> flush()
  after
    0 -> :ok
  end
end

receive do
  {:priority, msg} -> handle_priority(msg)
after
  0 -> :no_priority_messages
end

receive do
  {:normal, msg} -> handle_normal(msg)
after
  0 -> :no_normal_messages
end

defmodule PriorityQueue do
  def receive_message do
    receive do
      {:high, msg} -> {:high, msg}
    after
      0 ->
        receive do
          {:medium, msg} -> {:medium, msg}
        after
          0 ->
            receive do
              {:low, msg} -> {:low, msg}
            after
              100 -> :timeout
            end
        end
    end
  end
end

iex(1)> pid1 = spawn(fn -> loop() end)
iex(2)> pid2 = spawn(fn -> loop() end)

iex(3)> Process.link(pid1)
true

iex(1)> spawn_link(fn -> raise "Error!" end)
** (RuntimeError) Error!

iex(1)> Process.flag(:trap_exit, true)
false

iex(2)> spawn_link(fn -> raise "Error!" end)
#PID<0.123.0>

iex(3)> receive do
...>   msg -> IO.inspect(msg)
...> end
{:EXIT, #PID<0.123.0>, %RuntimeError{message: "Error!"}}

iex(1)> Process.exit(pid, :kill)
true

iex(2)> Process.exit(pid, :normal)
true

iex(1)> {pid, ref} = spawn_monitor(fn ->
...>   receive do
...>     _ -> :ok
...>   end
...> end)

iex(2)> send(pid, :stop)

iex(3)> receive do
...>   {:DOWN, ^ref, :process, ^pid, reason} ->
...>     IO.puts("Process exited: #{reason}")
...> end
Process exited: normal

iex(1)> pid = spawn(fn -> loop() end)
iex(2)> ref1 = Process.monitor(pid)
iex(3)> ref2 = Process.monitor(pid)

iex(1)> Process.demonitor(ref)
true

iex(2)> Process.demonitor(ref, [:flush])
true

iex(1)> pid = spawn(fn -> loop() end)
iex(2)> Process.register(pid, :my_server)
true

iex(3)> send(:my_server, :hello)
:hello

iex(1)> Process.whereis(:my_server)
#PID<0.123.0>

iex(2)> Process.whereis(:unknown)
nil

iex(1)> Process.registered()
[:my_server, :code_server, :file_server, ...]

iex(1)> Node.connect(:"node2@192.168.1.2")
true

iex(2)> Node.list()
[:"node2@192.168.1.2"]

iex(1)> send({:my_server, :"node2@192.168.1.2"}, :hello)
:hello

iex(1)> Node.spawn(:"node2@192.168.1.2", fn -> IO.puts("Remote!") end)

iex(1)> :pg2.create(:my_group)
:ok

iex(2)> :pg2.join(:my_group, self())
:ok

iex(3)> :pg2.get_members(:my_group)
[#PID<0.1.0>]

iex(4)> :pg2.broadcast(:my_group, :hello)
:hello

iex(1)> :pg.start_link()
iex(2)> :pg.join(:my_scope, :my_group, self())
iex(3)> :pg.get_members(:my_scope, :my_group)

defmodule Server do
  def call(pid, request) do
    ref = make_ref()
    send(pid, {self(), ref, request})

    receive do
      {^ref, response} -> response
    after
      5000 -> {:error, :timeout}
    end
  end

  def reply({from, ref}, response) do
    send(from, {ref, response})
  end
end

defmodule PubSub do
  def start do
    spawn(fn -> loop(%{}) end)
  end

  defp loop(subscribers) do
    receive do
      {:subscribe, pid, topic} ->
        new_subs = Map.update(subscribers, topic, [pid], &[pid | &1])
        loop(new_subs)

      {:unsubscribe, pid, topic} ->
        new_subs = Map.update(subscribers, topic, [], &List.delete(&1, pid))
        loop(new_subs)

      {:publish, topic, msg} ->
        Enum.each(Map.get(subscribers, topic, []), fn pid ->
          send(pid, msg)
        end)
        loop(subscribers)
    end
  end
end