庄周梦蝶

生活、程序、未来
   :: 首页 ::  ::  :: 聚合  :: 管理

Rabbitmq的网络层浅析

Posted on 2009-11-29 12:00 dennis 阅读(11558) 评论(8)  编辑  收藏 所属分类: erlang源码解读
    最近在锋爷的建议下开始读rabbitmq的源码,锋爷说这个项目已经很成熟,并且代码也很有借鉴和学习的意义,在自己写erlang代码之前看看别人是怎么写的,可以少走弯路,避免养成一些不好的习惯,学习一些最佳实践。读了一个星期,这个项目果然非常棒,代码也写的非常清晰易懂,一些细节的处理上非常巧妙,比如我这里想分享的网络层一节。
    Rabbitmq是一个MQ系统,也就是消息中间件,它实现了AMQP 0.8规范,简单来说就是一个TCP的广播服务器。AMQP协议,你可以类比JMS,不过JMS仅仅是java领域内的API规范,而AMQP比JMS更进一步,它有自己的wire-level protocol,有一套可编程的协议,中立于语言。简单介绍了Rabbitmq之后,进入正题。
    Rabbitmq充分利用了Erlang的分布式、高可靠性、并发等特性,首先看它的一个结构图:


这张图展现了Rabbitmq的主要组件和组件之间的关系,具体到监控树的结构,我画了一张图:







    顶层是rabbit_sup supervisor,它至少有两个子进程,一个是rabbit_tcp_client_sup,用来监控每个connection的处理进程 rabbit_reader的supervisor;rabbit_tcp_listener_sup是监控tcp_listener和 tcp_acceptor_sup的supervisor,tcp_listener里启动tcp服务器,监听端口,并且通过tcp_acceptor_sup启动N个tcp_accetpor,tcp_acceptor发起accept请求,等待客户端连接;tcp_acceptor_sup负责监控这些acceptor。这张图已经能给你一个大体的印象。
   
    讲完大概,进入细节,说说几个我觉的值的注意的地方:
1、tcp_accepto.erl,r对于accept采用的是异步方式,利用prim_inet:async_accept/2方 法,此模块没有被文档化,是otp库内部使用,通常来说没必要使用这一模块,gen_tcp:accept/1已经足够,不过rabbitmq是广播程 序,因此采用了异步方式。使用async_accept,需要打patch,以使得socket好像我们从gen_tcp:accept/1得到的一样:

handle_info({inet_async, LSock, Ref, {ok, Sock}},
            State = #state{callback={M,F,A}, sock=LSock, ref=Ref}) ->
    %%这里做了patch
    %% patch up the socket so it looks like one we got from
    %% gen_tcp:accept/1
    {ok, Mod} = inet_db:lookup_socket(LSock),
    inet_db:register_socket(Sock, Mod),

    try
        %% report
        {Address, Port}         = inet_op(fun () -> inet:sockname(LSock) end),
        {PeerAddress, PeerPort} = inet_op(fun () -> inet:peername(Sock) end),
        error_logger:info_msg("accepted TCP connection on ~s:~p from ~s:~p~n",
                              [inet_parse:ntoa(Address), Port,
                               inet_parse:ntoa(PeerAddress), PeerPort]),
        %% 调用回调模块,将Sock作为附加参数
        apply(M, F, A ++ [Sock])
    catch {inet_error, Reason} ->
            gen_tcp:close(Sock),
            error_logger:error_msg("unable to accept TCP connection: ~p~n",
                                   [Reason])
    end,

    %% 继续发起异步调用
    case prim_inet:async_accept(LSock, -1) of
        {ok, NRef} -> {noreply, State#state{ref=NRef}};
        Error -> {stop, {cannot_accept, Error}, none}
    end;
%%处理错误情况
handle_info({inet_async, LSock, Ref, {error, closed}},
            State=#state{sock=LSock, ref=Ref}) ->
    %% It would be wrong to attempt to restart the acceptor when we
    %% know this will fail.
    {stop, normal, State};

2、rabbitmq内部是使用了多个并发acceptor,这在高并发下、大量连接情况下有效率优势,类似java现在的nio框架采用多个reactor类似,查看tcp_listener.erl:

init({IPAddress, Port, SocketOpts,
      ConcurrentAcceptorCount, AcceptorSup,
      {M,F,A} = OnStartup, OnShutdown, Label}) ->
    process_flag(trap_exit, true),
    case gen_tcp:listen(Port, SocketOpts ++ [{ip, IPAddress},
                                             {active, false}]) of
        {ok, LSock} ->
             %%创建ConcurrentAcceptorCount个并发acceptor
            lists:foreach(fun (_) ->
                                  {ok, _APid} = supervisor:start_child(
                                                  AcceptorSup, [LSock])
                          end,
                          lists:duplicate(ConcurrentAcceptorCount, dummy)),

            {ok, {LIPAddress, LPort}} = inet:sockname(LSock),
            error_logger:info_msg("started ~s on ~s:~p~n",
                                  [Label, inet_parse:ntoa(LIPAddress), LPort]),
            %%调用初始化回调函数
            apply(M, F, A ++ [IPAddress, Port]),
            {ok, #state{sock = LSock,
                        on_startup = OnStartup, on_shutdown = OnShutdown,
                        label = Label}};
        {error, Reason} ->
            error_logger:error_msg(
              "failed to start ~s on ~s:~p - ~p~n",
              [Label, inet_parse:ntoa(IPAddress), Port, Reason]),
            {stop, {cannot_listen, IPAddress, Port, Reason}}
    end.

这里有一个技巧,如果要循环N次执行某个函数F,可以通过lists:foreach结合lists:duplicate(N,dummy)来处理。

lists:foreach(fun(_)-> F() end,lists:duplicate(N,dummy)).

3、simple_one_for_one策略的使用,可以看到对于tcp_client_sup和tcp_acceptor_sup都采用了simple_one_for_one策略,而非普通的one_fo_one,这是为什么呢?
这牵扯到simple_one_for_one的几个特点:
1)simple_one_for_one内部保存child是使用dict,而其他策略是使用list,因此simple_one_for_one更适合child频繁创建销毁、需要大量child进程的情况,具体来说例如网络连接的频繁接入断开。
2)使用了simple_one_for_one后,无法调用terminate_child/2 delete_child/2 restart_child/2

3)start_child/2 对于simple_one_for_one来说,不必传入完整的child spect,传入参数list,会自动进行参数合并在一个地方定义好child spec之后,其他地方只要start_child传入参数即可启动child进程,简化child都是同一类型进程情况下的编程

在 rabbitmq中,tcp_acceptor_sup的子进程都是tcp_acceptor进程,在tcp_listener中是启动了 ConcurrentAcceptorCount个tcp_acceptor子进程,通过supervisor:start_child/2方法:

%%创建ConcurrentAcceptorCount个并发acceptor
            lists:foreach(fun (_) ->
                                  {ok, _APid} = supervisor:start_child(
                                                  AcceptorSup, [
LSock])
                          end,
                          lists:duplicate(ConcurrentAcceptorCount, dummy)),

注意到,这里调用的start_child只传入了LSock一个参数,另一个参数CallBack是在定义child spec的时候传入的,参见tcp_acceptor_sup.erl:
init(Callback) ->
    {ok, {{simple_one_for_one, 10, 10},
          [{tcp_acceptor, {tcp_acceptor, start_link, [Callback]},
            transient, brutal_kill, worker, [tcp_acceptor]}]}}.

Erlang内部自动为simple_one_for_one做了参数合并,最后调用的是tcp_acceptor的init/2:

init({Callback, LSock}) ->
    case prim_inet:async_accept(LSock, -1) of
        {ok, Ref} -> {ok, #state{callback=Callback, sock=LSock, ref=Ref}};
        Error -> {stop, {cannot_accept, Error}}
    end.

对于tcp_client_sup的情况类似,tcp_client_sup监控的子进程都是rabbit_reader类型,在 rabbit_networking.erl中启动tcp_listenner传入的处理connect事件的回调方法是是 rabbit_networking:start_client/1:

start_tcp_listener(Host, Port) ->
    start_listener(Host, Port, "TCP Listener",
                   %回调的MFA
                   {?MODULE, start_client, []}).

start_client(Sock) ->
    {ok, Child} = supervisor:start_child(rabbit_tcp_client_sup, []),
    ok = rabbit_net:controlling_process(Sock, Child),
    Child ! {go, Sock},
    Child.

start_client调用了supervisor:start_child/2来动态启动rabbit_reader进程。

4、协议的解析,消息的读取这部分也非常巧妙,这一部分主要在rabbit_reader.erl中,对于协议的解析没有采用gen_fsm,而是实现了一个巧妙的状态机机制,核心代码在mainloop/4中:
%启动一个连接
start_connection(Parent, Deb, ClientSock) ->
    process_flag(trap_exit, true),
    {PeerAddressS, PeerPort} = peername(ClientSock),
    ProfilingValue = setup_profiling(),
    try
        rabbit_log:info("starting TCP connection ~p from ~s:~p~n",
                        [self(), PeerAddressS, PeerPort]),
         %延时发送握手协议
        Erlang:send_after(?HANDSHAKE_TIMEOUT * 1000, self(),
                          handshake_timeout),
        %进入主循环,更换callback模块,魔法就在这个switch_callback
        mainloop(Parent, Deb, switch_callback(
                                #v1{sock = ClientSock,
                                    connection = #connection{
                                      user = none,
                                      timeout_sec = ?HANDSHAKE_TIMEOUT,
                                      frame_max = ?FRAME_MIN_SIZE,
                                      vhost = none},
                                    callback = uninitialized_callback,
                                    recv_ref = none,
                                    connection_state = pre_init},
                                %%注意到这里,handshake就是我们的回调模块,8就是希望接收的数据长度,AMQP协议头的八个字节。
                                handshake, 8))

魔法就在switch_callback这个方法上:
switch_callback(OldState, NewCallback, Length) ->
    %发起一个异步recv请求,请求Length字节的数据
    Ref = inet_op(fun () -> rabbit_net:async_recv(
                              OldState#v1.sock, Length, infinity) end),
    %更新状态,替换ref和处理模块
    OldState#v1{callback = NewCallback,
                recv_ref = Ref}.


异步接收Length个数据,如果有,erlang会通知你处理。处理模块是什么概念呢?其实就是一个状态的概念,表示当前协议解析进行到哪一步,起一个label的作用,看看mainloop/4中的应用:

mainloop(Parent, Deb, State = #v1{sock= Sock, recv_ref = Ref}) ->
    %%?LOGDEBUG("Reader mainloop: ~p bytes available, need ~p~n", [HaveBytes, WaitUntilNBytes]),
    receive
        %%接收到数据,交给handle_input处理,注意handle_input的第一个参数就是callback
        {inet_async, Sock, Ref, {ok, Data}} ->
            %handle_input处理
            {State1, Callback1, Length1} =
                handle_input(State#v1.callback, Data,
                             State#v1{recv_ref = none}),

            %更新回调模块,再次发起异步请求,并进入主循环
            mainloop(Parent, Deb,
                     switch_callback(State1, Callback1, Length1));


handle_input有多个分支,每个分支都对应一个处理模块,例如我们刚才提到的握手协议:

%handshake模块,注意到第一个参数,第二个参数就是我们得到的数据
handle_input(handshake, <<"AMQP",1,1,ProtocolMajor,ProtocolMinor>>,
             State = #v1{sock = Sock, connection = Connection}) ->
     %检测协议是否兼容
    case check_version({ProtocolMajor, ProtocolMinor},
                       {?PROTOCOL_VERSION_MAJOR, ?PROTOCOL_VERSION_MINOR}) of
        true ->
            {ok, Product} = application:get_key(id),
            {ok, Version} = application:get_key(vsn),
            %兼容的话,进入connections start,协商参数
            ok = send_on_channel0(
                   Sock,
                   #'connection.start'{
                     version_major = ?PROTOCOL_VERSION_MAJOR,
                     version_minor = ?PROTOCOL_VERSION_MINOR,
                     server_properties =
                     [{list_to_binary(K), longstr, list_to_binary(V)} ||
                         {K, V} <-
                             [{"product",     Product},
                              {"version",     Version},
                              {"platform",    "Erlang/OTP"},
                              {"copyright",   ?COPYRIGHT_MESSAGE},
                              {"information", ?INFORMATION_MESSAGE}]],
                     mechanisms = <<"PLAIN AMQPLAIN">>,
                     locales = <<"en_US">> }),
            {State#v1{connection = Connection#connection{
                                     timeout_sec = ?NORMAL_TIMEOUT},
                      connection_state = starting},
             frame_header, 7};
         %否则,断开连接,返回可以接受的协议
        false ->
            throw({bad_version, ProtocolMajor, ProtocolMinor})
    end;

    其他协议的处理也是类似,通过动态替换callback的方式来模拟状态机做协议的解析和数据的接收,真的很巧妙!让我们体会到Erlang的魅力,FP的魅力。

5、序列图:
1)tcp server的启动过程:

2)一个client连接上来的处理过程:


    小结:从上面的分析可以看出,rabbitmq的网络层是非常健壮和高效的,通过层层监控,对每个可能出现的风险点都做了考虑,并且利用了prim_net模块做异步IO处理。分层也是很清晰,将业务处理模块隔离到client_sup监控下的子进程,将网络处理细节和业务逻辑分离。在协议的解析和业务处理上虽然没有采用gen_fsm,但是也实现了一套类似的状态机机制,通过动态替换Callback来模拟状态的变迁,非常巧妙。如果你要实现一个tcp server,强烈推荐从rabbitmq中扣出这个网络层,你只需要实现自己的业务处理模块即可拥有一个高效、健壮、分层清晰的TCP服务器。

评论

# re: Rabbitmq的网络层浅析[未登录]  回复  更多评论   

2009-11-29 12:48 by M
很有用,最近也在看,谢谢

# re: Rabbitmq的网络层浅析[未登录]  回复  更多评论   

2009-11-29 13:03 by M
prim_inet确实没有文档,我都没看明白,trapexit里有人问过这个问题,也有人(好像是erlang otp项目的人)说:
prim_inet is not documented because it is not intended to be a
supported/stable API which we will keep backwards compatible.

见:
http://www.trapexit.org/forum/viewtopic.php?p=29157

我有个疑问,按照simple_one_for_one的文档,supervisor:start_child每次将使用定义supervisor时init方法返回的的child spec,那么是不是说这种模式下每次只能有一个child,因为定义supervsor时,已指定了child ID,希望不吝赐教,谢谢。

# re: Rabbitmq的网络层浅析  回复  更多评论   

2009-11-29 13:21 by dennis

prim_inet,按照余锋老大的说法是可以用的,基本上接口不会有大的变更,gen_tcp其实是基于prim_net实现的。

使用simple_one_for_one,可以有多个child的,只不过这些child的是同一种类型的,看supervisor.erl的源码就知道,内部是动态保存在一个dict结构里dynamics = ?DICT:new(),因此是可保存多个:

{ok, Pid} ->
NState = State#state{dynamics =
?DICT:store(Pid, Args, State#state.dynamics)},
{reply, {ok, Pid}, NState};

这跟其他类型不一样:
%先判断是否存在name的child
case get_child(Child#child.name, State) of
false ->
case do_start_child(State#state.name, Child) of
{ok, Pid} ->
Children = State#state.children,
{{ok, Pid},
%加入到list
State#state{children =
[Child#child{pid = Pid}|Children]}};

# re: Rabbitmq的网络层浅析  回复  更多评论   

2009-11-29 13:22 by dennis
简单一句话,simple_one_for_one是是依据pid来保存的,而其他策略是依据child.name来保存的。

# re: Rabbitmq的网络层浅析[未登录]  回复  更多评论   

2009-11-29 17:23 by M
明白了,太感谢了!

# re: Rabbitmq的网络层浅析[未登录]  回复  更多评论   

2014-03-03 11:35 by 菲戈
更正一个小错误哈:
3.2中介绍说:使用了simple_one_for_one后,无法调用terminate_child/2 delete_child/2 restart_child/2

我看文档上说了,simple_one_for_one是可以使用 terminate_child/2的,但参数必须是进程的Pid。

-----------------------------------
不过这篇文章写得真的很详细啊,我解开了而我好多疑惑,赞一个!

# re: Rabbitmq的网络层浅析  回复  更多评论   

2014-09-16 17:21 by jj
看不到图片

# re: Rabbitmq的网络层浅析  回复  更多评论   

2015-03-20 02:09 by noboby
图挂了,分析得不错,但是图显示不了

只有注册用户登录后才能发表评论。


网站导航: