-module(redis_proxy). -include_lib("kernel/include/logger.hrl"). %% API -export([ start_link/1, start_link/2, stop/0, stop/1, status/0, status/1 ]). -export([ q/1, q/2, q/3, qp/1, qp/2, qp/3, foreach_redis/4, foreach_redis/5 ]). %% 利用redis自旋锁 -export([trans/2]). -define(SERVER, ?MODULE). -define(DEFAULT_POOL_SIZE, 16). -define(TIMEOUT, 5000). -define(REDIS_LOCK_TIME, 5). -define(REDIS_TIMEOUT, 5000). -define(TRANS_SLEEP_TIME_MAX, 100). start_link(Args) -> start_link(?SERVER, Args). start_link(Name, Args) when is_atom(Name), is_map(Args) -> Config = #{worker_module => redis_worker, name => {local, Name}, size => ?DEFAULT_POOL_SIZE}, M = maps:merge(Config, Args), poolboy:start_link(maps:to_list(M)); start_link(Name, Args) when is_list(Args) -> start_link(Name, maps:from_list(Args)). stop() -> stop(?SERVER). stop(Name) when is_atom(Name) -> poolboy:stop(Name). status() -> status(?SERVER). status(Name) when is_atom(Name) -> poolboy:status(Name). q(Command) -> q(Command, ?TIMEOUT). q(Command, Timeout) -> q(?SERVER, Command, Timeout). q(Name, Command, Timeout) -> poolboy:transaction(Name, fun(PID) -> eredis:q(PID, Command) end, Timeout). qp(Command) -> qp(Command, ?TIMEOUT). qp(Command, Timeout) -> qp(?SERVER, Command, Timeout). qp(Name, Command, Timeout) -> poolboy:transaction(Name, fun(PID) -> eredis:qp(PID, Command) end, Timeout). foreach_redis(REG, Fun, CntPerTime, Timeout) -> foreach_redis(?SERVER, REG, Fun, CntPerTime, Timeout). foreach_redis(Name, REG, Fun, CntPerTime, Timeout) -> foreach_redis(Name, <<"0">>, 0, REG, CntPerTime, Timeout, Fun). foreach_redis(_Name, <<"0">>, Times, _REG, _CNT, _Timeout, _Fun) when Times > 0 -> ok; foreach_redis(Name, Cur, Times, REG, CNT, Timeout, Fun) -> CMD = case REG == "*" of true -> ["SCAN", Cur, "COUNT", erlang:integer_to_list(CNT)]; false -> ["SCAN", Cur, "MATCH", REG, "COUNT", erlang:integer_to_list(CNT)] end, case q(Name, CMD, Timeout) of {ok, [Cur2, []]} -> foreach_redis(Name, Cur2, Times + 1, REG, CNT, Timeout, Fun); {ok, [Cur2, L]} -> case Fun of {M, F} -> M:F(L); _ -> Fun(L) end, foreach_redis(Name, Cur2, Times + 1, REG, CNT, Timeout, Fun) end. %% ==================== %% Fun = {M,F,A} | fun/0 | {fun/x,Args} trans(LockName, Fun) when is_binary(LockName) -> RandBin = erlang:term_to_binary(erlang:make_ref()), trans_i(LockName, RandBin, ?REDIS_LOCK_TIME, Fun). trans_i(LockKey, RandBin, LockTime, Fun) -> case redis_proxy:q(["SET", LockKey, RandBin, "EX", erlang:integer_to_list(LockTime), "NX"], ?REDIS_TIMEOUT) of {ok, <<"OK">>} -> Ret = case Fun of {M, F, A} -> catch apply(M, F, A); _ when is_function(Fun, 0) -> catch Fun(); {F, A} when is_function(F) -> catch erlang:apply(F, A) end, case redis_proxy:q(["GET", LockKey], ?REDIS_TIMEOUT) of {ok, RandBin} -> redis_proxy:q(["DEL", LockKey], ?REDIS_TIMEOUT); _ -> pass end, Ret; {ok, undefined} -> sleep_random(?TRANS_SLEEP_TIME_MAX), trans_i(LockKey, RandBin, LockTime, Fun); {error, Msg} -> ?LOG_ERROR("trans_error ~ts ~p", [LockKey, Msg]), sleep_random(?TRANS_SLEEP_TIME_MAX), trans_i(LockKey, RandBin, LockTime, Fun) end. sleep_random(Max) -> Sleep = rand:uniform(Max), timer:sleep(Sleep).