Erlang进程链接 #

一、链接概述 #

链接是进程间的双向关系，当一个进程退出时，链接的进程会收到EXIT信号。

二、link函数 #

2.1 创建链接 #

erlang

-module(link_basic).
-export([demo/0]).

demo() ->
    process_flag(trap_exit, true),
    Pid = spawn_link(fun worker/0),
    io:format("Linked to: ~p~n", [Pid]),
    receive
        {'EXIT', Pid, Reason} ->
            io:format("Process exited: ~p~n", [Reason])
    end.

worker() ->
    timer:sleep(1000),
    exit(normal).

2.2 spawn_link #

erlang

-module(spawn_link_demo).
-export([demo/0]).

demo() ->
    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.

2.3 unlink #

erlang

-module(unlink_demo).
-export([demo/0]).

demo() ->
    Pid = spawn_link(fun() -> timer:sleep(5000) end),
    timer:sleep(100),
    unlink(Pid),
    io:format("Unlinked from: ~p~n", [Pid]).

三、EXIT信号 #

3.1 EXIT信号类型 #

信号	说明
`normal`	正常退出
`shutdown`	关闭信号
`{shutdown, Term}`	带原因的关闭
`kill`	强制杀死
其他	异常退出

3.2 捕获EXIT #

erlang

-module(trap_exit).
-export([demo/0]).

demo() ->
    process_flag(trap_exit, true),
    Pid = spawn_link(fun() ->
        timer:sleep(1000),
        exit(crashed)
    end),
    receive
        {'EXIT', Pid, Reason} ->
            io:format("Caught exit: ~p~n", [Reason])
    end.

3.3 不捕获EXIT #

erlang

-module(no_trap_exit).
-export([demo/0]).

demo() ->
    spawn_link(fun() ->
        timer:sleep(1000),
        exit(crashed)
    end),
    timer:sleep(2000).

不捕获EXIT时，收到异常退出信号会导致当前进程也退出。

3.4 kill信号 #

erlang

-module(kill_signal).
-export([demo/0]).

demo() ->
    process_flag(trap_exit, true),
    Pid = spawn_link(fun() ->
        receive
            _ -> ok
        end
    end),
    exit(Pid, kill),
    receive
        {'EXIT', Pid, killed} ->
            io:format("Process killed~n")
    end.

kill信号无法被捕获，会强制终止进程。

四、monitor函数 #

4.1 创建监控 #

erlang

-module(monitor_basic).
-export([demo/0]).

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

4.2 spawn_monitor #

erlang

-module(spawn_monitor_demo).
-export([demo/0]).

demo() ->
    {Pid, Ref} = spawn_monitor(fun() ->
        timer:sleep(1000),
        exit(normal)
    end),
    receive
        {'DOWN', Ref, process, Pid, Reason} ->
            io:format("Process down: ~p~n", [Reason])
    end.

4.3 demonitor #

erlang

-module(demonitor_demo).
-export([demo/0]).

demo() ->
    Pid = spawn(fun() -> timer:sleep(5000) end),
    Ref = monitor(process, Pid),
    demonitor(Ref),
    io:format("Demonitored~n").

4.4 demonitor选项 #

erlang

-module(demonitor_options).
-export([demo/0]).

demo() ->
    Pid = spawn(fun() -> exit(normal) end),
    Ref = monitor(process, Pid),
    
    demonitor(Ref, [flush]),
    io:format("Demonitored with flush~n").

选项：

flush：清除可能已发送的DOWN消息
info：返回监控是否存在

五、link vs monitor #

5.1 比较表 #

特性	link	monitor
方向	双向	单向
多次监控	不支持	支持
主动取消	unlink	demonitor
信号类型	EXIT	DOWN
适用场景	相关进程	监控进程

5.2 使用场景 #

erlang

-module(link_vs_monitor).
-export([with_link/0, with_monitor/0]).

with_link() ->
    process_flag(trap_exit, true),
    Pid = spawn_link(fun worker/0),
    receive
        {'EXIT', Pid, Reason} -> {exited, Reason}
    end.

with_monitor() ->
    {Pid, Ref} = spawn_monitor(fun worker/0),
    receive
        {'DOWN', Ref, process, Pid, Reason} -> {down, Reason}
    end.

worker() ->
    timer:sleep(1000),
    exit(normal).

六、容错模式 #

6.1 简单监督者 #

erlang

-module(simple_supervisor).
-export([start/1, loop/1]).

start(ChildSpec) ->
    spawn(?MODULE, loop, [ChildSpec]).

loop(ChildSpec) ->
    process_flag(trap_exit, true),
    Pid = start_child(ChildSpec),
    receive
        {'EXIT', Pid, _Reason} ->
            io:format("Child exited, restarting~n"),
            loop(ChildSpec)
    end.

start_child({M, F, A}) ->
    spawn_link(M, F, A).

6.2 重启策略 #

erlang

-module(restart_strategy).
-export([start/2, loop/2]).

start(ChildSpec, Strategy) ->
    spawn(?MODULE, loop, [ChildSpec, Strategy]).

loop(ChildSpec, permanent) ->
    process_flag(trap_exit, true),
    Pid = start_child(ChildSpec),
    receive
        {'EXIT', Pid, _Reason} ->
            io:format("Restarting permanent child~n"),
            loop(ChildSpec, permanent)
    end;
loop(ChildSpec, transient) ->
    process_flag(trap_exit, true),
    Pid = start_child(ChildSpec),
    receive
        {'EXIT', Pid, normal} ->
            io:format("Transient child exited normally~n");
        {'EXIT', Pid, _Reason} ->
            io:format("Restarting transient child~n"),
            loop(ChildSpec, transient)
    end;
loop(ChildSpec, temporary) ->
    Pid = start_child(ChildSpec),
    receive
        {'EXIT', Pid, Reason} ->
            io:format("Temporary child exited: ~p~n", [Reason])
    end.

start_child({M, F, A}) ->
    spawn_link(M, F, A).

6.3 心跳监控 #

erlang

-module(heartbeat_monitor).
-export([start/1, loop/2]).

start(Interval) ->
    spawn(?MODULE, loop, [Interval, undefined]).

loop(Interval, Pid) when is_pid(Pid) ->
    receive
        {'DOWN', _Ref, process, Pid, Reason} ->
            io:format("Monitored process died: ~p~n", [Reason]),
            loop(Interval, undefined)
    after Interval ->
        Pid ! ping,
        receive
            pong -> loop(Interval, Pid)
        after Interval ->
            io:format("Heartbeat timeout~n"),
            loop(Interval, undefined)
        end
    end;
loop(Interval, undefined) ->
    receive
        {monitor, NewPid} ->
            Ref = monitor(process, NewPid),
            io:format("Monitoring: ~p~n", [NewPid]),
            loop(Interval, NewPid)
    end.

七、实际应用 #

7.1 连接管理器 #

erlang

-module(connection_manager).
-export([start/0, connect/1, disconnect/1, loop/1]).

start() ->
    spawn(?MODULE, loop, [#{}]).

connect(Pid) ->
    connection_manager ! {connect, Pid},
    ok.

disconnect(Pid) ->
    connection_manager ! {disconnect, Pid},
    ok.

loop(Connections) ->
    receive
        {connect, Pid} ->
            Ref = monitor(process, Pid),
            loop(Connections#{Pid => Ref});
        {disconnect, Pid} ->
            case maps:find(Pid, Connections) of
                {ok, Ref} ->
                    demonitor(Ref),
                    loop(maps:remove(Pid, Connections));
                error ->
                    loop(Connections)
            end;
        {'DOWN', Ref, process, Pid, Reason} ->
            io:format("Connection lost: ~p (~p)~n", [Pid, Reason]),
            loop(maps:remove(Pid, Connections))
    end.

7.2 进程池 #

erlang

-module(worker_pool).
-export([start/1, get_worker/0, return_worker/1, loop/2]).

start(Size) ->
    Workers = [spawn_worker() || _ <- lists:seq(1, Size)],
    spawn(?MODULE, loop, [Workers, #{}]).

spawn_worker() ->
    spawn_link(fun worker_loop/0).

get_worker() ->
    worker_pool ! {get, self()},
    receive
        {worker, Pid} -> {ok, Pid}
    after 5000 -> {error, timeout}
    end.

return_worker(Pid) ->
    worker_pool ! {return, Pid},
    ok.

loop(Available, Busy) ->
    receive
        {get, From} when Available =/= [] ->
            [Pid | Rest] = Available,
            From ! {worker, Pid},
            loop(Rest, Busy#{Pid => true});
        {return, Pid} ->
            loop([Pid | Available], maps:remove(Pid, Busy));
        {'EXIT', Pid, _Reason} ->
            NewAvailable = lists:delete(Pid, Available),
            NewBusy = maps:remove(Pid, Busy),
            NewWorker = spawn_worker(),
            loop([NewWorker | NewAvailable], NewBusy)
    end.

worker_loop() ->
    receive
        {work, From, Task} ->
            Result = do_work(Task),
            From ! {result, Result},
            worker_loop()
    end.

do_work(T) -> T.

八、总结 #

本章学习了：

link函数
EXIT信号
monitor函数
link vs monitor
容错模式
实际应用