Erlang，尝试制作 gen_server: 调用有很多响应

Question

尝试在项目中使用 OTP-style 并得到一个 OTP-interface 问题。哪种方案更受欢迎/更美观？

我有什么：

1. 带有mochiweb 的网络服务器
2. 一个进程，产生许多 (1000-2000) 个子进程。 儿童包含状态（netflow-speed）。如果需要，处理给孩子的代理消息并创建新的孩子。

在 mochiweb 中，我有一页显示所有演员的速度，乳清是如何制作的：

    nf_collector ! {get_abonents_speed, self()},
    receive
        {abonents_speed_count, AbonentsCount} ->
            ok
    end,
%% write http header, chunked
%% and while AbonentsCount != 0,  receive speed and write http

这不是选择风格，我怎么能理解。解决方案：

1. 在 API 同步函数中获取所有速度的请求并返回所有速度的列表。但我想马上写给客户。

2. API-function 的一个参数是回调：

   nf_collector:get_all_speeds(fun (Speed) -> Resp:write_chunk(templater(Speed)) end)
   

3. 返回迭代器： get_all_speeds 的结果之一将与接收块一起使用。每次调用都会返回{ok, Speed}，最后返回{end}。

get_all_speeds() ->
    nf_collector ! {get_abonents_speed, self()},
    receive
        {abonents_speed_count, AbonentsCount} ->
            ok
    end,
    {ok, fun() -> 
        create_receive_fun(AbonentsCount)
    end}.

create_receive_fun(0)->
    {end};

create_receive_fun(Count)->
        receive
            {abonent_speed, Speed} ->
                Speed
        end,
        {ok, Speed, create_receive_fun(Count-1)}.

Answer 1

从主管那里生出你的“孩子”：

-module(ch_sup).
-behaviour(supervisor).
-export([start_link/0, init/1, start_child/1]).
start_link() -> supervisor:start_link({local, ?MODULE}, ?MODULE, []).
init([]) -> {ok, {{simple_one_for_one}, [{ch, {ch, start_link, []}, transient, 1000, worker, [ch]}]}}.
start_child(Data) -> supervisor:start_child(?MODULE, [Data]).

用 ch_sup:start_child/1 启动它们（数据是什么）。

将您的孩子实现为 gen_server：

-module(ch).
-behaviour(gen_server).
-record(?MODULE, {speed}).

...

get_speed(Pid, Timeout) ->
    try
        gen_server:call(Pid, get, Timeout)
    catch
        exit:{timeout, _} -> timeout;
        exit:{noproc, _} -> died
    end
.

...

handle_call(get, _From, St) -> {reply, {ok, St#?MODULE.speed}, St} end.

您现在可以使用主管获取正在运行的孩子的列表并查询他们，但您必须接受在获取孩子列表和调用他们之间有孩子死亡的可能性，显然孩子可能出于某种原因活着但没有响应，或响应错误等。

上面的 get_speed/2 函数返回 {ok, Speed} 或 dead 或 timeout。您可以根据自己的应用需求进行适当的过滤；通过列表理解很容易，这里有一些。

只是速度：

[Speed || {ok, Speed} <- [ch:get_speed(Pid, 1000) || Pid <-
    [Pid || {undefined, Pid, worker, [ch]} <-
        supervisor:which_children(ch_sup)
        ]
    ]].

Pid 和速度元组：

[{Pid, Speed} || {Pid, {ok, Speed}} <-
    [{Pid, ch:get_speed(Pid, 1000)} || Pid <-
        [Pid || {undefined, Pid, worker, [ch]} <-
                supervisor:which_children(ch_sup)]
        ]
    ].

所有结果，包括超时和在您到达之前死亡的孩子的“死亡”结果：

[{Pid, Any} || {Pid, Any} <-
    [{Pid, ch:get_speed(Pid, 1000)} || Pid <-
        [Pid || {undefined, Pid, worker, [ch]} <-
                supervisor:which_children(ch_sup)]
        ]
    ].

在大多数情况下，您几乎可以肯定除了速度之外不想要任何东西，因为您将如何处理死亡和超时？你希望那些死去的人由主管重生，所以问题在你知道的时候或多或少地得到解决，而超时，就像任何错误一样，是一个单独的问题，可以用你看到的任何方式处理适合...不过，无需将故障修复逻辑与数据检索逻辑混合在一起。

现在，我认为您在帖子中提到的所有这些问题，但我不太确定，1000 的超时是针对每个调用的，并且每个调用是一个接一个地同步的，因此对于 1000 个超时时间为 1 秒的孩子，可能需要 1000 秒才能产生任何结果。让时间超时 1ms 可能是答案，但要正确地做到这一点有点复杂：

get_speeds() ->
    ReceiverPid = self(),
    Ref = make_ref(),
    Pids = [Pid || {undefined, Pid, worker, [ch]} <-
            supervisor:which_children(ch_sup)],
    lists:foreach(
        fun(Pid) -> spawn(
            fun() -> ReceiverPid ! {Ref, ch:get_speed(Pid, 1000)} end
            ) end,
        Pids),
    receive_speeds(Ref, length(Pids), os_milliseconds(), 1000)
.

receive_speeds(_Ref, 0, _StartTime, _Timeout) ->
    [];
receive_speeds(Ref, Remaining, StartTime, Timeout) ->
    Time = os_milliseconds(),
    TimeLeft = Timeout - Time + StartTime,
    receive
        {Ref, acc_timeout} ->
            [];
        {Ref, {ok, Speed}} ->
            [Speed | receive_speeds(Ref, Remaining-1, StartTime, Timeout)];
        {Ref, _} ->
            receive_speeds(Ref, Remaining-1, StartTime, Timeout)
    after TimeLeft ->
        []
    end
.

os_milliseconds() ->
    {OsMegSecs, OsSecs, OsMilSecs} = os:timestamp(),
    round(OsMegSecs*1000000 + OsSecs + OsMilSecs/1000)
.

这里每个调用都在不同的进程中产生并收集回复，直到“主超时”或全部收到。

代码大部分是从我的各种作品中剪切和粘贴的，并通过手动和搜索替换进行编辑，以匿名并删除多余的，所以它可能主要是可编译的质量，但我不保证我没有不要破坏任何东西。