Erlang 并发

简介

Erlang 基于轻量级进程和消息传递的并发模型，支持构建高度可扩展的系统。进程彼此隔离，不共享内存，通过异步消息进行通信。该模型消除了共享内存系统中常见的并发缺陷。

BEAM 虚拟机能够高效地调度数百万个进程，每个进程拥有独立的堆和邮箱。进程创建快速且开销极小，使得“每个实体一个进程”（process per entity）的设计模式成为可能。链接（links）和监控器（monitors）提供故障检测能力，而选择性接收（selective receive）则支持灵活的消息处理模式。

本技能涵盖：进程创建与派生（spawning）、消息传递模式、进程链接与监控、选择性接收、错误传播、并发设计模式，以及可扩展并发系统的构建。

进程创建与派生

为并发任务执行创建轻量级进程。

%% 基础进程派生
simple_spawn() ->
    Pid = spawn(fun() ->
        io:format("Hello from process ~p~n", [self()])
    end),
    Pid.

%% 带参数的派生
spawn_with_args(Message) ->
    spawn(fun() ->
        io:format("Message: ~p~n", [Message])
    end).

%% 派生并注册
spawn_registered() ->
    Pid = spawn(fun() -> loop() end),
    register(my_process, Pid),
    Pid.

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

%% 派生并链接（linked processes）
spawn_linked() ->
    spawn_link(fun() ->
        timer:sleep(1000),
        io:format("Linked process done~n")
    end).

%% 派生并监控
spawn_monitored() ->
    {Pid, Ref} = spawn_monitor(fun() ->
        timer:sleep(500),
        exit(normal)
    end),
    {Pid, Ref}.

%% 进程池
create_pool(N) ->
    [spawn(fun() -> worker_loop() end) || _ <- lists:seq(1, N)].

worker_loop() ->
    receive
        {work, Data, From} ->
            Result = process_data(Data),
            From ! {result, Result},
            worker_loop();
        stop ->
            ok
    end.

process_data(Data) -> Data * 2.

%% 并行 map
pmap(F, List) ->
    Parent = self(),
    Pids = [spawn(fun() ->
        Parent ! {self(), F(X)}
    end) || X <- List],
    [receive {Pid, Result} -> Result end || Pid <- Pids].


%% Fork-join 模式
fork_join(Tasks) ->
    Self = self(),
    Pids = [spawn(fun() ->
        Result = Task(),
        Self ! {self(), Result}
    end) || Task <- Tasks],
    [receive {Pid, Result} -> Result end || Pid <- Pids].

轻量级进程以极低开销实现海量并发。

消息传递模式

进程之间通过异步消息传递进行通信，不依赖共享内存。

%% 发送与接收
send_message() ->
    Pid = spawn(fun() ->
        receive
            {From, Msg} ->
                io:format("Received: ~p~n", [Msg]),
                From ! {reply, "Acknowledged"}
        end
    end),
    Pid ! {self(), "Hello"},
    receive
        {reply, Response} ->
            io:format("Response: ~p~n", [Response])
    after 5000 ->
        io:format("Timeout~n")
    end.

%% 请求-响应模式
request(Pid, Request) ->
    Ref = make_ref(),
    Pid ! {self(), Ref, Request},
    receive
        {Ref, Response} -> {ok, Response}
    after 5000 ->
        {error, timeout}
    end.

server_loop() ->
    receive
        {From, Ref, {add, A, B}} ->
            From ! {Ref, A + B},
            server_loop();
        {From, Ref, {multiply, A, B}} ->
            From ! {Ref, A * B},
            server_loop();
        stop -> ok
    end.

%% 发布-订阅模式
start_pubsub() ->
    spawn(fun() -> pubsub_loop([]) end).

pubsub_loop(Subscribers) ->
    receive
        {subscribe, Pid} ->
            pubsub_loop([Pid | Subscribers]);
        {unsubscribe, Pid} ->
            pubsub_loop(lists:delete(Pid, Subscribers));
        {publish, Message} ->
            [Pid ! {message, Message} || Pid <- Subscribers],
            pubsub_loop(Subscribers)
    end.

%% 流水线模式
pipeline(Data, Functions) ->
    lists:foldl(fun(F, Acc) -> F(Acc) end, Data, Functions).

concurrent_pipeline(Data, Stages) ->
    Self = self(),
    lists:foldl(fun(Stage, AccData) ->
        Pid = spawn(fun() ->
            Result = Stage(AccData),
            Self ! {result, Result}
        end),
        receive {result, R} -> R end
    end, Data, Stages).

消息传递可在无锁前提下实现安全的并发通信。

链接与监控

链接在进程间建立双向连接；监控则提供单向观测能力。

%% 进程链接
link_example() ->
    process_flag(trap_exit, true),
    Pid = spawn_link(fun() ->
        timer:sleep(1000),
        exit(normal)
    end),
    receive
        {'EXIT', Pid, Reason} ->
            io:format("Process exited: ~p~n", [Reason])
    end.

%% 监控
monitor_example() ->
    Pid = spawn(fun() ->
        timer:sleep(500),
        exit(normal)
    end),
    Ref = monitor(process, Pid),
    receive
        {'DOWN', Ref, process, Pid, Reason} ->
            io:format("Process down: ~p~n", [Reason])
    end.

%% 监督者模式
supervisor() ->
    process_flag(trap_exit, true),
    Worker = spawn_link(fun() -> worker() end),
    supervisor_loop(Worker).

supervisor_loop(Worker) ->
    receive
        {'EXIT', Worker, _Reason} ->
            NewWorker = spawn_link(fun() -> worker() end),
            supervisor_loop(NewWorker)
    end.

worker() ->
    receive
        crash -> exit(crashed);
        work -> worker()
    end.

链接与监控机制可支撑容错系统的构建，并实现自动故障检测。

最佳实践

1. 可自由创建进程——因其轻量且派生开销极小

2. 仅使用消息传递进行进程间通信，避免共享状态

3. 在 receive 中设置合理超时，防止因失败导致无限阻塞

4. 当无需双向链接时，优先使用监控实现单向观测

5. 尽量减少进程状态，以降低每个进程的内存占用

6. 谨慎使用注册名称——全局名称会限制系统可扩展性

7. 结合链接与监控实现完善的错误处理，保障系统容错能力

8. 使用选择性接收，按需处理特定消息，其余消息保留在邮箱中排队

9. 避免消息堆积——确保 receive 子句覆盖所有可能的消息类型

10. 对并发系统进行性能分析，识别瓶颈并优化热点路径

常见陷阱

1. 创建进程过少——未能充分利用 Erlang 的并发模型优势

2. receive 中未设超时——故障发生时将导致无限阻塞

3. 邮箱中消息持续堆积——引发内存泄漏与性能下降

4. 将共享 ETS 表用作互斥锁替代方案——破坏进程隔离性优势

5. 未处理全部消息类型——导致无法匹配的消息不断积压，造成邮箱溢出

6. 监督者中忘记 trap exits——致使错误处理机制失效

7. 建立循环链接——缺乏适当监督时易引发级联故障

8. 用进程实现细粒度并行——引入额外开销却无实际收益

9. 未监控派生的进程——无法感知其失败状态

10. 过度使用注册名称——形成单点故障与竞争瓶颈

适用场景

当并发任务需要隔离性和独立状态时，应使用进程。

在分布式系统中，所有进程间通信均应采用消息传递方式。

利用链接与监控构建具备容错能力的监督层级结构。

为并发请求处理与并行计算创建进程池。

针对复杂消息协议，使用选择性接收实现精细化控制。

参考资料

• Erlang Concurrency Tutorial
• Learn You Some Erlang — Concurrency
• Erlang Process Manual
• Designing for Scalability with Erlang/OTP
• Programming Erlang