(注:分析代码基于RabbitMQ 2.8.2)
网络层的启动也是作为上一篇文章中提到的一个启动步骤来启动的,入口为[$RABBIT_SRC/src/rabbit_networking.erl --> boot/0],代码如下:
boot() ->
ok = start(),
ok = boot_tcp(),
ok = boot_ssl().[$RABBIT_SRC/src/rabbit_networking.erl-->start/0]构建rabbit_client_sup子监控树,[$RABBIT_SRC/src/rabbit_networking.erl --> boot_tcp/0],启动TCP监听,并构建成tcp_listener_sup子监控树,这两个子树都属于rabbit最顶层监控结点rabbit_sup(SSL的类似)。最终的监控树如下:

(注:方框代表supervisor类型,圆角代表worker;虚线灰底的文字代表相应层子进程重启策略;各结点以各自所在的模块命名,实际监控树中的结点名有部分不同。)
[$RABBIT_SRC/src/tcp_listener.erl --> init/1]通过gen_tcp:listen/2启动监听,并根据ConcurrentAcceptorCount参数启动指定数量个tcp_acceptor(当前这个数量是1)。
[$RABBIT_SRC/src/tcp_acceptor.erl]通过prim_inet:async_accept/2开始异步接受来自客户端的连接,主要代码如下:
handle_cast(accept, State) ->
ok = file_handle_cache:obtain(),
accept(State);
handle_info({inet_async, LSock, Ref, {ok, Sock}},
State = #state{callback={M,F,A}, sock=LSock, ref=Ref}) ->
%% 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),
%% handle
file_handle_cache:transfer(apply(M, F, A ++ [Sock])),
ok = file_handle_cache:obtain(),
%% accept more
accept(State);
accept(State = #state{sock=LSock}) ->
case prim_inet:async_accept(LSock, -1) of
{ok, Ref} -> {noreply, State#state{ref=Ref}};
Error -> {stop, {cannot_accept, Error}, State}
end.[$RABBIT_SRC/src/ file_handle_cache.erl]在这里的主要作用是管理每个进程拥有的文件描述符数量,确保每个进程不会超过设定的上限。其中,obtain/0方法增加拥有的文件描述符数量,transfer/1将文件描述符的拥有权转移到另外一个进程(原进程拥有的打开描述符数量减1,转移目标进程拥有的文件描述符数量加1,这里的目标进程是通过callback返回的rabbit_reader进程)。
state结构中的callback是由[$RABBIT_SRC/src/rabbit_networking.erl]在启动TCP监听端口时指定的,其值为[$RABBIT_SRC/src/rabbit_networking.erl --> start_client/1],每个客户端连接上来后,会通过apply方法调用(见上面代码中的apply(M, F, A)),实际调用[$RABBIT_SRC/src/rabbit_networking.erl --> start_client/2],主要代码如下:
start_client(Sock, SockTransform) ->
{ok, _Child, Reader} = supervisor:start_child(rabbit_tcp_client_sup, []),
ok = rabbit_net:controlling_process(Sock, Reader),
Reader ! {go, Sock, SockTransform},
Reader.(这里rabbit_tcp_client_sup对应上图中rabbit_client_sup结点)
supervisor:start_child(rabbit_tcp_client_sup, [])会按照上图中rabbit_client_sup下的监控树结构依次构造各个子结点。也就是说在每一次服务端接受一个新的连接时,都会创建一个rabbit_client_sup的子进程(simple_one_for_one),并依次构建各个子结点(上图中黄色区域)。并最终返回创建的rabbit_reader。
rabbit_net:controlling_process/2将上一步返回的rabbit_reader进程作为接收socket消息的进程(调用gen_tcp:controlling_process/2实现,原始接受消息的是tcp_acceptor进程)。
Reader ! {go, Sock, SockTransform}向目标rabbit_reader进程发送消息,指示开始接受从客户端发送来的数据。
rabbit_reader进程在创建后,会通过如下代码等待上述的go消息:
receive
{go, Sock, SockTransform} ->
start_connection(
Parent, ChannelSupSupPid, Collector, StartHeartbeatFun, Deb, Sock,
SockTransform)
end(参见[$RABBIT_SRC/src/rabbit_reader.erl --> init/4])
收到go消息后,rabbit_reader开始与客户端进行消息交互。(参见[$RABBIT_SRC/src/rabbit_reader.erl --> start_connection /7])主要代码如下:
start_connection(Parent, ChannelSupSupPid, Collector, StartHeartbeatFun, Deb,
Sock, SockTransform) ->
process_flag(trap_exit, true),
ConnStr = name(Sock),
log(info, "accepting AMQP connection ~p (~s)~n", [self(), ConnStr]),
ClientSock = socket_op(Sock, SockTransform),
erlang:send_after(?HANDSHAKE_TIMEOUT * 1000, self(),
handshake_timeout),
State = #v1{parent = Parent,…}, // 初始化state,部分代码省略
try
recvloop(Deb, switch_callback(rabbit_event:init_stats_timer(
State, #v1.stats_timer),
handshake, 8)),
log(info, "closing AMQP connection ~p (~s)~n", [self(), ConnStr])
catch
// 异常处理
after
rabbit_net:maybe_fast_close(ClientSock),
rabbit_event:notify(connection_closed, [{pid, self()}])
end,
done.从上面的log语句可以看出,这时候与客户端的连接已经确定,要开始AMQP的握手过程。接收消息的主循环由recvloop/2来完成,其中第二个参数State保存当前连接的各种状态。recvloop按照AMQP协议对数据进行解析。
recvloop并没有使用任何一个OTP的行为模式,而是自己设计的一种状态转换的逻辑,该逻辑的实现依赖于AMQP协议本身,其中关键的数据是State里的callback以及recv_len。AMQP协议要求,客户端建立一个连接时,必需向服务端发送8个字节的协议头(包含AMQP4个字符以及客户端使用的AMQP版本),所以rabbit在一开始的时候,设定的recv_len=8,也就是说rabbit希望从客户端收到8个字节的数据,然后对这8个字节的处理对应handshake处理逻辑(见上面代码的recvloop/2调用)。handshake完成后,rabbit会根据AMQP协议定义的frame格式等待客户端发送来的数据:先是7个字符的frame头(recv_len=7, callback=frame_header),然后是长度声明在头中的负载数据(recv_len=PayloadSize+1,callback=frame_payload,每个frame都有个结束符,所以这里负载长度要加1),收到负载数据后,会根据负载数据做对应处理,处理完成后,继续待续下一个frame。所以整个协议解析的过程就如下图所示:

rabbit会根据客户端发送的请求进行相应的操作,比如创建channel,创建exchange,创建queue等等。下篇会分析在创建每个实体时,都会做哪些工作。