Erlang 分布式机制

简介

Erlang 内置的分布式机制支持跨多个节点构建集群化、容错型系统。不同节点上的进程可通过与本地通信完全一致的消息传递原语进行透明通信。这种位置透明性使分布式编程变得自然且直观。

分布式层自动处理网络通信、序列化及节点连接管理。节点通过名称相互发现，进程可通过全局注册名或 pid 引用被任意访问。深入理解分布式模式，对构建可扩展、高弹性的系统至关重要。

本技能涵盖节点连接与集群组建、分布式消息传递、全局名称注册、分布式监督（supervision）、网络分区处理、RPC 模式，以及生产级分布式应用的构建方法。

节点连接

节点通过相互连接组成集群，以实现分布式计算与容错能力。

%% 启动具名节点
%% erl -name node1@hostname -setcookie secret
%% erl -sname node2 -setcookie secret

%% 连接节点
connect_nodes() ->
    Node1 = 'node1@host',
    Node2 = 'node2@host',
    net_kernel:connect_node(Node2).

%% 查看已连接节点
list_nodes() ->
    Nodes = [node() | nodes()],
    io:format("Connected nodes: ~p~n", [Nodes]).

%% 监控节点连接状态
monitor_nodes() ->
    net_kernel:monitor_nodes(true),
    receive
        {nodeup, Node} ->
            io:format("Node up: ~p~n", [Node]);
        {nodedown, Node} ->
            io:format("Node down: ~p~n", [Node])
    end.

%% 节点配置
start_distributed() ->
    {ok, _} = net_kernel:start([mynode, shortnames]),
    erlang:set_cookie(node(), secret_cookie).

%% 隐藏节点（用于监控）
connect_hidden(Node) ->
    net_kernel:connect_node(Node),
    erlang:disconnect_node(Node),
    net_kernel:hidden_connect_node(Node).

%% 获取节点信息
node_info() ->
    #{
        name => node(),
        cookie => erlang:get_cookie(),
        nodes => nodes(),
        alive => is_alive()
    }.

节点连接能力支持构建具备自动发现功能的分布式集群。

分布式消息传递

使用与本地消息发送完全相同的语法，向远程节点上的进程发送消息。

%% 向远程节点上已注册的进程发送消息
send_remote(Node, Name, Message) ->
    {Name, Node} ! Message.

%% 在远程节点上启动进程
spawn_on_remote(Node, Fun) ->
    spawn(Node, Fun).

spawn_on_remote(Node, Module, Function, Args) ->
    spawn(Node, Module, Function, Args).

%% 分布式请求-响应模式
remote_call(Node, Module, Function, Args) ->
    Pid = spawn(Node, fun() ->
        Result = apply(Module, Function, Args),
        receive
            {From, Ref} -> From ! {Ref, Result}
        end
    end),
    Ref = make_ref(),
    Pid ! {self(), Ref},
    receive
        {Ref, Result} -> {ok, Result}
    after 5000 ->
        {error, timeout}
    end.

%% 分布式任务分发
-module(work_dispatcher).
-export([start/0, dispatch/1]).

start() ->
    register(?MODULE, spawn(fun() -> loop([]) end)).

dispatch(Work) ->
    ?MODULE ! {dispatch, Work}.

loop(Workers) ->
    receive
        {dispatch, Work} ->
            Node = select_node(nodes()),
            Pid = spawn(Node, fun() -> do_work(Work) end),
            loop([{Pid, Node} | Workers])
    end.

select_node(Nodes) ->
    lists:nth(rand:uniform(length(Nodes)), Nodes).

do_work(Work) ->
    Result = process_work(Work),
    io:format("Work done on ~p: ~p~n", [node(), Result]).

process_work(Work) -> Work * 2.

%% 远程 group leader（用于输出重定向）
remote_process_with_io(Node) ->
    spawn(Node, fun() ->
        group_leader(self(), self()),
        io:format("Output from ~p~n", [node()])
    end).

位置透明的消息机制实现了无缝的分布式通信。

全局名称注册

在分布式集群中跨节点全局注册进程名称。

%% 全局注册
register_global(Name) ->
    Pid = spawn(fun() -> global_loop() end),
    global:register_name(Name, Pid),
    Pid.

global_loop() ->
    receive
        {From, Message} ->
            From ! {reply, Message},
            global_loop();
        stop -> ok
    end.

%% 向全局注册的进程发送消息
send_global(Name, Message) ->
    case global:whereis_name(Name) of
        undefined ->
            {error, not_found};
        Pid ->
            Pid ! Message,
            ok
    end.

%% 带冲突解决机制的全局注册
register_with_resolve(Name) ->
    Pid = spawn(fun() -> server_loop() end),
    ResolveFun = fun(Name, Pid1, Pid2) ->
        %% 保留节点名字典序更小的进程
        case node(Pid1) < node(Pid2) of
            true -> Pid1;
            false -> Pid2
        end
    end,
    global:register_name(Name, Pid, ResolveFun).

server_loop() ->
    receive
        Message ->
            io:format("Received: ~p on ~p~n", [Message, node()]),
            server_loop()
    end.

%% 全局同步
sync_global() ->
    global:sync().

%% 列出所有全局注册的名称
list_global_names() ->
    global:registered_names().

%% 节点重连后确保全局注册
ensure_global_registration(Name, Fun) ->
    case global:whereis_name(Name) of
        undefined ->
            Pid = spawn(Fun),
            global:register_name(Name, Pid),
            Pid;
        Pid ->
            Pid
    end.

全局注册机制支持与位置无关的进程发现。

分布式监督

跨多个节点监督进程，实现集群范围内的容错能力。

-module(distributed_supervisor).
-behaviour(supervisor).

-export([start_link/0, start_worker/1]).
-export([init/1]).

start_link() ->
    supervisor:start_link({local, ?MODULE}, ?MODULE, []).

start_worker(Node) ->
    ChildSpec = #{
        id => make_ref(),
        start => {worker, start_link, [Node]},
        restart => permanent,
        type => worker
    },
    supervisor:start_child(?MODULE, ChildSpec).

init([]) ->
    SupFlags = #{
        strategy => one_for_one,
        intensity => 5,
        period => 60
    },
    {ok, {SupFlags, []}}.

%% 在指定节点上启动的工作进程模块
-module(worker).
-export([start_link/1, loop/0]).

start_link(Node) ->
    Pid = spawn_link(Node, ?MODULE, loop, []),
    {ok, Pid}.

loop() ->
    receive
        stop -> ok;
        Msg ->
            io:format("Worker on ~p: ~p~n", [node(), Msg]),
            loop()
    end.

%% 分布式进程组示例
-module(pg_example).
-export([start/0, join/1, broadcast/1]).

start() ->
    pg:start_link().

join(Group) ->
    pg:join(Group, self()).

broadcast(Group, Message) ->
    Members = pg:get_members(Group),
    [Pid ! Message || Pid <- Members].

分布式监督机制可在节点故障时持续保障系统健康运行。

RPC 与远程执行

以多种调用模式在远程节点上执行函数调用。

%% 基础 RPC
simple_rpc(Node, Module, Function, Args) ->
    rpc:call(Node, Module, Function, Args).

%% 带超时的 RPC
timed_rpc(Node, Module, Function, Args, Timeout) ->
    rpc:call(Node, Module, Function, Args, Timeout).

%% 异步 RPC
async_rpc(Node, Module, Function, Args) ->
    Key = rpc:async_call(Node, Module, Function, Args),
    %% 后续获取结果
    rpc:yield(Key).

%% 并行 RPC（多节点）
parallel_rpc(Nodes, Module, Function, Args) ->
    rpc:multicall(Nodes, Module, Function, Args).

%% 带超时的并行 RPC（返回结果）
parallel_rpc_results(Nodes, Module, Function, Args) ->
    rpc:multicall(Nodes, Module, Function, Args, 5000).

%% Cast（即发即弃）
cast_rpc(Node, Module, Function, Args) ->
    rpc:cast(Node, Module, Function, Args).

%% 广播至所有节点
broadcast_rpc(Module, Function, Args) ->
    Nodes = [node() | nodes()],
    rpc:multicall(Nodes, Module, Function, Args).

%% 在节点间并行映射（pmap）
pmap_nodes(Fun, List) ->
    Nodes = nodes(),
    DistFun = fun(X) ->
        Node = lists:nth((X rem length(Nodes)) + 1, Nodes),
        rpc:call(Node, erlang, apply, [Fun, [X]])
    end,
    lists:map(DistFun, List).

RPC 提供了便捷的远程执行能力，并保持位置透明性。

网络分区与 CAP 理论

处理网络分区问题，并理解 CAP 定理中的权衡取舍。

%% 检测网络分区
detect_partition() ->
    ExpectedNodes = [node1@host, node2@host, node3@host],
    CurrentNodes = nodes(),
    Missing = ExpectedNodes -- CurrentNodes,
    case Missing of
        [] -> ok;
        Nodes -> {partition, Nodes}
    end.

%% 分区恢复策略
-module(partition_handler).
-export([monitor_cluster/1]).

monitor_cluster(ExpectedNodes) ->
    net_kernel:monitor_nodes(true),
    monitor_loop(ExpectedNodes, nodes()).

monitor_loop(Expected, Current) ->
    receive
        {nodeup, Node} ->
            NewCurrent = [Node | Current],
            case length(NewCurrent) == length(Expected) of
                true ->
                    io:format("Cluster fully connected~n"),
                    heal_partition();
                false ->
                    ok
            end,
            monitor_loop(Expected, NewCurrent);

        {nodedown, Node} ->
            NewCurrent = lists:delete(Node, Current),
            io:format("Partition detected: ~p~n", [Node]),
            monitor_loop(Expected, NewCurrent)
    end.

heal_partition() ->
    %% 分区恢复后同步状态
    global:sync(),
    ok.

%% 基于多数派的共识机制
-module(consensus).
-export([propose/2, vote/3]).

propose(Nodes, Value) ->
    Ref = make_ref(),
    [Node ! {vote, self(), Ref, Value} || Node <- Nodes],
    collect_votes(Ref, length(Nodes), 0).

collect_votes(_Ref, Total, Votes) when Votes > Total div 2 ->
    {ok, majority};
collect_votes(_Ref, Total, Total) ->
    {error, no_majority};
collect_votes(Ref, Total, Votes) ->
    receive
        {vote, Ref, accept} ->
            collect_votes(Ref, Total, Votes + 1);
        {vote, Ref, reject} ->
            collect_votes(Ref, Total, Votes)
    after 5000 ->
        {error, timeout}
    end.

vote(From, Ref, Value) ->
    Decision = evaluate_proposal(Value),
    From ! {vote, Ref, Decision}.

evaluate_proposal(_Value) -> accept.

分区处理策略可在网络故障期间维持系统可用性。

最佳实践

1. 本地集群使用短名称（short names），广域网部署使用长名称（long names）

2. 在可信集群的所有节点上设置相同 cookie，以保障安全性

3. 监控节点连接状态，及时检测并应对网络分区

4. 谨慎使用全局注册，因其会引入额外的协调开销

5. 实现分区检测与恢复策略，提升系统韧性

6. 面向最终一致性进行设计，以适配分布式系统固有的 CAP 局限性

7. 简单调用优先使用 RPC，复杂协议则推荐基于消息传递实现

8. 使用 toxiproxy 或混沌工程等工具模拟网络故障进行测试

9. 为所有分布式调用设置合理超时，以应对网络延迟或拥塞

10. 采用分布式监督机制，保障跨节点的容错能力

常见陷阱

1. 未设置 cookie：导致节点无法连接，引发静默失败

2. 过度依赖全局注册表：造成单点故障与性能瓶颈

3. 忽略节点断连处理：导致进程无限挂起

4. 假设网络绝对可靠：在网络分区时引发逻辑错误

5. RPC 调用使用过长超时：故障期间引发级联延迟

6. 未对网络分区场景开展测试：遗漏关键失效路径

7. 分区恢复后忘记同步全局注册表

8. 多台机器使用相同节点名：引发命名冲突

9. 未监控节点健康状态：无法及时识别集群退化

10. 在分布式环境中强求严格一致性：违背 CAP 定理基本约束

适用场景

当构建需要高可用性与容错能力的系统时，应启用分布式机制。

对需自动故障转移的关键服务，应采用分布式监督机制。

当单机资源已达瓶颈，需借助水平扩展提升吞吐能力时，应利用多节点架构。

当地理分布本身能带来显著延迟优势时，应构建分布式系统。

为实现负载在多台服务器之间的均衡分发，应启用集群能力。

构建具备韧性的微服务架构时，应应用各类分布式模式。

参考资料

• Erlang Distribution Protocol
• Distributed Erlang
• Learn You Some Erlang - Distributed OTP
• Designing Distributed Systems
• CAP Theorem Explained