<8>redis及erl-redis阅读

1   redis的功能相当的强大，里面的发布订阅pub/sub和设计模式中的观察者模式很相似
pub/sub不仅仅解决发布者和订阅者直接代码级别耦合也解决两者在物理部署上的耦合。
并且发布和订阅功能都是多对多的
见发布订阅的例子 http://bbs.chinaunix.net/thread-3755278-1-1.html

erl-redis是litaocheng实现的一个精炼有简单易懂的redis客户端
里面有很多巧妙的地方， 着也是一个标准的erlang客户端实现，如果想实现客户端用这种方法还是很不错的

其实整个实现流程是很清晰易懂的：
从redis_conn_sup:start_link()开始启动监督进程，然后通过start_child启动子进程，
start_link() ->
    ?INFO2("start the supervisor", []),
    supervisor:start_link({local, ?CONN_SUP}, ?MODULE, []).


%% @doc the connection supervisor callback
init([]) -> 
    ?DEBUG2("init supervisor", []),
    Stragegy = {simple_one_for_one, 100000, 60},
    Client = {undefined, {redis_client, start_link, []},
                permanent, 1000, worker, [redis_client]},

    {ok, {Stragegy, [Client]}}.

redis_client.erl中实现start_link和基本的tcp逻辑

start_link(Host, Port, Passwd) ->
    ?DEBUG2("start_link redis_client ~p:~p", [Host, Port]),
    gen_server:start_link(?MODULE, {{Host, Port}, Passwd}, []).

-spec start_link(inet_host(), inet_port(), passwd(), atom()) ->
    {'ok', any()} | 'ignore' | {'error', any()}.
start_link(Host, Port, Passwd, Name) ->
    ?DEBUG2("start_link redis_client ~p", [Name]),
    gen_server:start_link({local, Name}, ?MODULE, {{Host, Port}, Passwd}, []).
%%             
%% gen_server callbacks
%%                 
init({Server = {Host, Port}, Passwd}) ->
    process_flag(trap_exit, true),
    case gen_tcp:connect(Host, Port, ?TCP_OPTS, ?CONN_TIMEOUT) of
        {ok, Sock} ->
            case do_auth(Sock, Passwd) of
                ok ->
                    Tid = do_create_table(),
                    {ok, #state{server = Server, sock = Sock, pubsub_tid = Tid}};
                {tcp_error, Reason} ->
                    {stop, Reason};
                _ ->
                    ?ERROR2("auth failed", []),
                    {stop, auth_failed}
            end;
        {error, Reason} ->
            {stop, Reason}
    end.

其中do_create_table()是创建一个ets表，用来存储订阅相关的数据存储;
do_create_table() ->
    ets:new(dummy, [set, private, {keypos, #pubsub.id},
            {read_concurrency, true}]).
关于ets的read_concurrency选项就是并发读取数据，这个选项默认是false

看这个tcp选项
生成连接redis server的tcp连接
这里的CONN_TIMEOUT 是5000 5秒钟
tcp_option是
-define(TCP_OPTS, [inet, binary, {active, once},
            {packet, line},
            {nodelay, false},
            {recbuf, 16#1000},
            {sndbuf, 16#10000},
            {send_timeout, 5000}, {send_timeout_close, true}]).


在调用redis的命令时候，执行：
%% call the command
call(Cmd) ->
    call(Cmd, ?NONE).
call(Cmd, Fun) ->
    case Pipeline of
        false ->
            % normal model
            R = redis_client:command(Client, Cmd),
            ?IF(Fun =/= ?NONE, Fun(R), R);
        true ->
            % pipeline model
            add_pipeline_cmd(Cmd, Fun)
    end.

if的定义
-define(IF(C, T, F), (case (C) of true -> (T); false -> (F) end)).
继续看command的调用
%% @doc send the command to redis server
-spec command(client(), iolist()) -> any().
command(Client, Data) ->
    call(Client, {command, {Data, ?COMMAND_TIMEOUT}}).
call(Client, Req) ->
    gen_server:call(Client, Req, infinity).



调用和接收处理
命令的调用是同步的，
handle_call({command, {Data, Timeout}}, _From,
        State = #state{sock = Sock, server = _Server, ctx = normal}) ->
    ?DEBUG2("command:~n~p~n\t=> ~p", [Data, _Server]),
    Reply = do_send_recv(Data, Sock, Timeout),
    {reply, Reply, State};

do_recv(Sock, PState, Timeout) ->
    receive
        {tcp, Sock, Packet} ->
            %?DEBUG2("receive packet :~p", [Packet]),
            do_handle_packet(Sock, Packet, PState, Timeout);
        {tcp_closed, _Socket} ->
            ?ERROR2("socket closed by remote peer", []),
            exit({error, tcp_closed});
        {tcp_error, _Socket, Reason} ->
            ?ERROR2("recv message error:~p", [Reason]),
            exit({error, {tcp_error, Reason}})
    after  
        Timeout ->
            ?ERROR2("recv message timeout", []),
            exit({error, recv_timeout})
    end.

总结下其中写的比较巧妙的地方:
(1)  在redis_cient.erl 中gen_server:start_link时候可以传进去name参数
Name的生成 ：
关于这个巧妙生成name的生成方法:
name(Host, Port, UserData) ->
    to_name(Host, Port, UserData, true).
existing_name(Host, Port, UserData) ->
    to_name(Host, Port, UserData, false).


to_name(Host, Port, UserData, First) when is_list(UserData);
                                is_atom(UserData);
                                is_integer(UserData) ->
    L = lists:concat(["redis_client_", Host, "_", Port, "_", UserData]),
    to_atom(L, First).

to_atom(String, true) ->
    list_to_atom(String);
to_atom(String, false) ->
    list_to_existing_atom(String).
只要输入自己的标识符UserData就可以生成名称

(2) 使代码更加简洁,把
关于redis.erl的模块，使用了一种比较少用的方法，
目的是为了使用起来更方便，没有其它的特殊用途，
刚好ligaoren对这种语法写了详细的博客说明，还提到本项目

-spec handler(client()) -> redis_handler().
-spec handler(client()) -> redis_handler().
handler(Client) ->
    redis:new(Client, false).
handler(Client) ->
    redis:new(Client, false).
http://www.cnblogs.com/me-sa/archive/2012/02/16/2354499.html

(3) 启动流程简单
README.md写的太好了，启动方法很简单，
其中有一种启动方法巧妙的启动了一个reids pool，这里默认是启动了5个redis进程
% in main supervisor:
    {redis_conn_sup, {redis_conn_sup, start_link, []},
        permanent, 1000, supervisor, [redis_client]}

    % start client pool
    [begin
        Name = redis_client:name(Host, Port, I),
        {ok, _} = redis_conn_sup:connect(Host, Port, Pass, Name)
    end || I <- lists:seq(1, 5)],

    % random select a client
    Selected = redis_client:existing_name(Host, Port, random:uniform(5)),
    Redis = redis_client:handler(Selected),
    Redis:set("k1", "v1"),
    Redis:get("k1").
还有第五种使用方法中，随机查找一个进程的方法
Redis = redis_conn_sup:sup_rand_client(Host, Port, Pass, Pool),
sup_rand_client() ->
    Children = supervisor:which_children(?CONN_SUP),
    Len = length(Children),
    {_Id, Child, worker, _Modules} = lists:nth(random:uniform(Len), Children),
    redis_client:handler(Child).

(4)这是一个标准的erlang客户端实现

erl-redis项目地址
https://github.com/litaocheng/erl-redis

评论

2 楼 ZacMa 2013-03-28  

哈哈，突然感觉里面没怎么介绍，全是贴代码了

1 楼 惊涛翻案 2013-03-28  

马博士，给我开课吧