如何让两个Elixir应用（服务端&客户端）通过Phoenix.PubSub通信？

核心原理 #

Phoenix PubSub支持分布式节点通信，只要两个Elixir应用运行在同一个Erlang集群内，就能共享PubSub的Topic。核心是让服务端与客户端加入同一Erlang集群，同时配置PubSub使用相同的实例名称和适配器。

一、服务端实现步骤 #

1. 添加依赖 #

在mix.exs中确保引入:phoenix_pubsub：

defp deps do
  [
    {:phoenix_pubsub, "~> 2.1"}
  ]
end

2. 配置PubSub实例 #

在config/config.exs中指定PubSub的名称与适配器（本地集群推荐用Phoenix.PubSub.PG2）：

config :my_server, MyServer.PubSub,
  name: MyServer.PubSub,
  adapter: Phoenix.PubSub.PG2

3. 启动PubSub服务 #

在lib/my_server/application.ex的启动函数中加入PubSub进程：

defmodule MyServer.Application do
  use Application

  @impl true
  def start(_type, _args) do
    children = [
      {Phoenix.PubSub, name: MyServer.PubSub}
    ]

    Supervisor.start_link(children, strategy: :one_for_one)
  end
end

4. 实现消息处理GenServer #

这个GenServer负责向:server_broadcast_topic广播消息，同时订阅:client_response_topic接收客户端反馈：

defmodule MyServer.MessageHandler do
  use GenServer

  @server_topic :server_broadcast_topic
  @client_topic :client_response_topic

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

  @impl true
  def init(_) do
    # 订阅客户端响应Topic
    Phoenix.PubSub.subscribe(MyServer.PubSub, @client_topic)
    # 启动定时广播（模拟业务消息）
    Process.send_after(self(), :broadcast_message, 5000)
    {:ok, nil}
  end

  @impl true
  def handle_info(:broadcast_message, state) do
    message = "Server broadcast at #{DateTime.utc_now()}"
    Phoenix.PubSub.broadcast(MyServer.PubSub, @server_topic, {:server_message, message})
    IO.puts("Server broadcasted: #{message}")
    # 继续定时广播
    Process.send_after(self(), :broadcast_message, 5000)
    {:noreply, state}
  end

  @impl true
  def handle_info({:client_message, message}, state) do
    IO.puts("Server received client message: #{message}")
    {:noreply, state}
  end
end

记得把这个GenServer加入application.ex的子进程列表：

children = [
  {Phoenix.PubSub, name: MyServer.PubSub},
  MyServer.MessageHandler
]

5. 启动服务端节点 #

启动时指定节点名称，方便客户端加入集群：

iex --sname server@localhost -S mix

二、客户端实现步骤 #

1. 添加依赖 #

同样在mix.exs中引入:phoenix_pubsub：

defp deps do
  [
    {:phoenix_pubsub, "~> 2.1"}
  ]
end

2. 配置PubSub #

客户端需指向服务端的PubSub实例名称，在config/config.exs中配置：

config :my_client, MyClient.PubSub,
  name: MyServer.PubSub, # 与服务端PubSub名称完全一致
  adapter: Phoenix.PubSub.PG2

3. 实现订阅响应GenServer #

这个GenServer订阅服务端的:server_broadcast_topic，收到消息后向:client_response_topic发送反馈：

defmodule MyClient.MessageSubscriber do
  use GenServer

  @server_topic :server_broadcast_topic
  @client_topic :client_response_topic

  def start_link(opts) do
    client_name = Keyword.get(opts, :name, __MODULE__)
    GenServer.start_link(__MODULE__, nil, name: client_name)
  end

  @impl true
  def init(_) do
    # 订阅服务端广播Topic
    Phoenix.PubSub.subscribe(MyServer.PubSub, @server_topic)
    {:ok, nil}
  end

  @impl true
  def handle_info({:server_message, message}, state) do
    IO.puts("Client #{inspect(self())} received: #{message}")
    # 发送响应到客户端Topic
    response = "Client #{inspect(self())} acknowledged: #{message}"
    Phoenix.PubSub.broadcast(MyServer.PubSub, @client_topic, {:client_message, response})
    {:noreply, state}
  end
end

4. 启动客户端并加入集群 #

打开新终端，启动客户端节点：

iex --sname client1@localhost -S mix

在客户端的IEx中连接服务端集群：

Node.connect(:"server@localhost")

验证连接成功：

Node.list() # 返回 [:"server@localhost"] 即为成功

可以启动多个客户端实例：

# 启动第一个客户端
MyClient.MessageSubscriber.start_link(name: MyClient.Client1)
# 启动第二个客户端
MyClient.MessageSubscriber.start_link(name: MyClient.Client2)

三、验证通信效果 #

1. 服务端每隔5秒广播消息，所有客户端会收到并打印内容
2. 每个客户端收到消息后会自动发送响应，服务端会打印所有客户端的反馈
3. 启动多个客户端时，每个实例都会独立接收和响应消息

关键注意事项 #

• 本地节点默认使用相同的Erlang Cookie，若跨机器部署需手动指定一致的--cookie <cookie>参数启动节点
• 服务端与客户端的PubSub实例名称必须完全一致
• 确保两个应用使用的Erlang/OTP版本兼容

内容的提问来源于stack exchange，提问作者Elmseld