Erlang:进程的简单pubsub - 我的方法好吗？

Question

免责声明:我对Erlang和OTP很新.

我想在Erlang/OTP中使用一个简单的pubsub,其中进程可以在某个"集线器"订阅并接收发送到该集线器的消息副本.

我知道gen_event,但它在一个事件管理器进程中处理事件,而我希望每个订阅者都是一个独立的自治进程.而且,我无法理解gen_event处理程序的监督.不幸的是,谷歌的结果充满了XMPP(Ejabberd)和RabbitMQ链接,所以我没有发现任何与我的想法相关的内容.

我的想法是这样的pubsub模型无缝映射到监督树.因此,我想扩展主管(gen_server引擎盖下),以便能够向其所有孩子发送演员信息.

我在快速而肮脏的自定义"调度程序"行为中攻击了这个:

-module(dispatcher).
-extends(supervisor).
-export([notify/2, start_link/2, start_link/3, handle_cast/2]).

start_link(Mod, Args) ->
    gen_server:start_link(dispatcher, {self, Mod, Args}, []).

start_link(SupName, Mod, Args) ->
    gen_server:start_link(SupName, dispatcher, {SupName, Mod, Args}, []).

notify(Dispatcher, Message) ->
    gen_server:cast(Dispatcher, {message, Message}).

handle_cast({message, Message}, State) ->
    {reply, Children, State} = supervisor:handle_call(which_children, dummy, State),
    Pids = lists:filter(fun(Pid) -> is_pid(Pid) end,
                 lists:map(fun({_Id, Child, _Type, _Modules}) -> Child end,
                           Children)),
    [gen_server:cast(Pid, Message) || Pid <- Pids],
    {noreply, State}.

然而,虽然一切看起来一切正常(儿童接收消息并在失败时无缝重启),但我想知道这是个好主意.

请某人批评(或批准)我的方法,和/或推荐一些替代方案吗？

Answer 1

我最近用gproc来实现pubsub.自述文件中的示例可以解决问题.

subscribe(EventType) ->
    %% Gproc notation: {p, l, Name} means {(p)roperty, (l)ocal, Name}
    gproc:reg({p, l, {?MODULE, EventType}}).

notify(EventType, Msg) ->
    Key = {?MODULE, EventType},
    gproc:send({p, l, Key}, {self(), Key, Msg}).

Answer 2

从你的代码中我看起来gen_event处理程序是完美的匹配.

从一个调度消息的中央进程调用处理程序回调,但这些回调不应该做太多工作.

因此,如果您需要一个具有自己的订阅者状态的自治进程,只需在事件回调中发送一条消息.

通常这些自​​治进程将是gen_servers,您只需从事件回调中调用gen_server:cast.

监督是一个单独的问题,可以通过OTP附带的通常监督基础设施来处理.您希望如何进行监督取决于订户进程的语义.如果它们都是相同的服务器,您可以使用simple_one_for_one例如.

在init订阅者进程的回调中,您可以将gen_event:add_handler添加它们的调用放入事件管理器.

如果您使用该gen_event:add_sup_handler函数添加您的进程(如果您的语义适合您),您甚至可以使用事件管理器作为主管.

更好地理解gen_event的在线资源:了解一些Erlang章节

否则,Erlang书籍都会有一些gen_event介绍.可能是Erlang和OTP in Action中最彻底的一个

哦,顺便说一句:我不会因此而砍掉你自己的主管.