erlang分布式入门（四）-TCP Server的详解与改进

在上一篇实现的erlang分布式入门（三）-TCP Server-Client 中的accept函数如下：

 

accept(LSocket) ->  
    {ok, Socket} = gen_tcp:accept(LSocket),  
    spawn(fun() -> loop(Socket) end),  
    accept(LSocket). 

 

使用BIF的spawn方法，创建了一个新的进程loop来处理客户端连接，主要业务在loop函数中实现，然后继续accept新的客户端连接。

spawn的说明如下：

spawn(Fun) -> pid()

Types:
Fun = function()

Returns the pid of a new process started by the application of Fun to the empty list []. Otherwise works like spawn/3.

spawn(Node, Fun) -> pid()

Types:
Node = node()
Fun = function()

Returns the pid of a new process started by the application of Fun to the empty list [] on Node. If Node does not exist, a useless pid is returned. Otherwise works like spawn/3.

spawn(Module, Function, Args) -> pid()

Types:
Module = Function = atom()
Args = [term()]

Returns the pid of a new process started by the application of Module:Function to Args. The new process created will be placed in the system scheduler queue and be run some time later.

error_handler:undefined_function(Module, Function, Args) is evaluated by the new process if Module:Function/Arity does not exist (where Arity is the length of Args). The error handler can be redefined (see process_flag/2). If error_handler is undefined, or the user has redefined the default error_handler its replacement is undefined, a failure with the reason undef will occur.

> spawn(speed, regulator, [high_speed, thin_cut]).
<0.13.1>

spawn(Node, Module, Function, Args) -> pid()

Types:
Node = node()
Module = module()
Function = atom()
Args = [term()]

Returns the pid of a new process started by the application of Module:Function to Args on Node. If Node does not exists, a useless pid is returned. Otherwise works like spawn/3.

 

但是由于gen_tcp:accept是阻塞的，所以我们在服务端启动tcp_server之后没有返回值，erlang的命令行一直阻塞在那里，一直等到有客户端连接上去才有反应。

以下是accept函数的说明：

accept(ListenSocket) -> {ok, Socket} | {error, Reason}
accept(ListenSocket, Timeout) -> {ok, Socket} | {error, Reason}

Types:
ListenSocket = socket()
Returned by listen/2.
Timeout = timeout()
Socket = socket()
Reason = closed | timeout | system_limit | inet:posix()

Accepts an incoming connection request on a listen socket. Socket must be a socket returned from listen/2. Timeout specifies a timeout value in ms, defaults to infinity.

Returns {ok, Socket} if a connection is established, or {error, closed} if ListenSocket is closed, or {error, timeout} if no connection is established within the specified time, or {error, system_limit} if all available ports in the Erlang emulator are in use. May also return a POSIX error value if something else goes wrong, see inet(3) for possible error values.

Packets can be sent to the returned socket Socket using send/2. Packets sent from the peer are delivered as messages:

{tcp, Socket, Data}

unless {active, false} was specified in the option list for the listen socket, in which case packets are retrieved by calling recv/2.

 

我们知道nginx的Master进程只用于接收客户端连接，然后把连接交给worker子进程去处理。这是高并发服务器处理大量连接的很有效的方式。

以下是改进版的TCP服务端：

 

echo_server.erl

-module(echo_server).

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

%% @spec start(Port::integer(), Max::integer())  
%% @doc echo server
start(Port, Max) ->
        generic_server:start(echo_server, Port, Max, {?MODULE, loop}).

%% @spec loop(Sock::port())
%% @doc echo_server

loop(Sock) ->
        case gen_tcp:recv(Sock, 0) of
                {ok, Data} ->
                        io:format("server recv data close~p~n",[Data]),
                        gen_tcp:send(Sock, Data),
                        loop(Sock);
                {error, closed} ->
                        io:format("client sock close~n"),
                        gen_server:cast(echo_server, {connect_close, self()})
        end.

 

generic_server.erl

%% a generic tcp server
-module(generic_server).
-author('cheng [e-mail]').
-vsn('0.1').
-behaviour(gen_server).

-define(TCP_OPTIONS, [binary, {packet, raw}, {active, false}, {reuseaddr, true}]).
-record(server_state, {
                        port,                   % listen port
                        loop,                   % the logic fun
                        ip=any,                 % ip
                        lsocket=null,           % listen socket
                        conn=0,                 % curent connect
                        maxconn                 % max connect
                        }).

%% internal export
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, code_change/3, terminate/2]).

%% start tcp server 
-export([start/4]).

-export([accept_loop/5]).

%% start the generic server
start(Name, Port, Max, Loop) ->
        State = #server_state{port=Port, loop=Loop, maxconn=Max},
        io:format("max connection is ~p~n", [Max]),
        gen_server:start_link({local, Name}, ?MODULE, State, []).

%% create listen socket
init(State = #server_state{port=Port}) ->
        case gen_tcp:listen(Port, ?TCP_OPTIONS) of
                {ok, LSocket} ->
                        {ok, accept(State#server_state{lsocket=LSocket})};
                {error, Reason} ->
                        {stop, {create_listen_socket, Reason}}
        end.


%% accept spawn a new process
accept(State =#server_state{lsocket=LSocket, loop=Loop, conn=Conn, maxconn=Max}) ->
        proc_lib:spawn(generic_server, accept_loop, [self(), LSocket, Loop, Conn, Max]),
        State.

%% accept the new connection
accept_loop(Server, LSocket, {M, F}, Conn, Max) ->
        {ok, Sock} = gen_tcp:accept(LSocket),
        if
                Conn + 1 > Max ->
                        io:format("reach the max connection~n"),
                        gen_tcp:close(Sock);
                true ->
                        gen_server:cast(Server, {accept_new, self()}),
                        M:F(Sock)
        end.

%% the server receive the notify that a connect has construct
handle_cast({accept_new, _Pid}, State=#server_state{conn=Cur}) ->
        io:format("current connect:~p~n", [Cur+1]),
        {noreply, accept(State#server_state{conn=Cur+1})};

%% someone connect has been close, so change the max connect
handle_cast({connect_close, _Pid}, State=#server_state{conn=Cur}) ->
        io:format("current connect:~p~n", [Cur-1]),
        {noreply, State#server_state{conn=Cur-1}}.

handle_call(_From, _Request, State) -> {noreply, State}.
handle_info(_Info, State) -> {noreply, State}.
code_change(_OldVsn, State, _Extra) -> {ok, State}.

terminate(_Reason, _State) -> ok.

 

主要使用了erlang的gen_server内置函数，详解：http://www.erlang.org/doc/design_principles/gen_server_concepts.html