为了避免系统过载, 对系统做负载保护, 往往需要对系统被调用次数做一定的限制, 比如一段时间内调用次数不能超过某个值.
先简化下场景, 让描述变得简单一些, 系统在任意60秒内只允许10次调用.
絮絮叨叨
有一种方案, 是初始化limit(10), 每次调用将limit减1, 每隔60秒, 将limit 重置为10. 这种方案能满足需求吗?
仅依靠上述方案, 可以满足60秒内限制10次调用, 但是并不能达到"任意"的效果.
一段时间内限制调用次数, 需要注意几点:
1, 活动因子
活动因子对应是否被允许调用, 一个活动因子对应一次调用是否被允许.
2, 活动因子的状态切换
活跃 <=> 冷却
活跃的活动因子表明这次调用是被允许的, 冷却的活动因子则表示此次调用不可进行.
3, 冷却时间
在某次调用被允许后, 一个活动因子就需要被冷却, 直到约定的时间后, 才能被重新激活为活跃状态.
4, 调用者等待超时
5, 等待队列深度
利用活动因子以及活动因子状态的切换, 也就是在系统中存在活跃的活动因子时, 即执行一次调用, 并在执行调用时, 将活动因子冷却. 若所有的活动因子都处于冷却状态, 则不能执行本次调用. 那么, 对于一个活动因子而言,在一段时间内只可能被调用一次. 在系统初始化是, 设置10个活动因子, 在每次调用时, 冷却一个活动因子, 并设置定时使其在60秒后恢复活跃状态. 这样就能达到"在任意60秒内只有10次调用"的效果了.
若现在没有可用的活动因子时, 一般来说，系统对调用者有两种处理方式:
1, 直接失败返回
2, 等待冷却的活动因子恢复
若等待冷却的活动因子恢复, 就需要维护一个waiting queue(很常见, 这在之前的pool 管理的相关随笔中都有提到), 将调用者写入等待队列, 并在冷却的活动因子恢复时, 对waiting queue 中等待的调用者做后续的相关操作. 如若调用者等待一段时间后, 取消等待操作, 需要在waiting queue 中将对应的等待信息移除.
当无数的(非常多的)调用者都在等待活动因子时, 就需要考量等待队列的深度了. 如果有100 个调用者都在等待活动因子, 但是最多只有10个活动因子, 也就是消化这100 次调用需要100/10 = 10 个等待周期(600 秒), 大多数系统场景中, 就已经没有意义了. 所以在等待队列深度到达一定值之后, 后来的调用者就不需要在等待了.
简单实现
絮叨了不少, 用Erlang简单实现下吧.
使用一个gen_server 进程, 用来维护活动因子, 计数, 活动因子的状态切换, 调用者等待队列维护.
record 定义:
1 -record(state, { 2 max_num = 10, 3 waiting_queue = queue:new(), 4 time_interval = 60000}).
60000, 10 代表60秒内允许10次调用. waiting_queue 用来保存等待的调用者.
get_token callback 方法:
1 handle_cast({get_token, OriginPid, MsgID, Msg}, 2 #state{max_num = 0, waiting_queue = WaitingQueue} = State) -> 3 erlang:monitor(process, OriginPid), 4 NewState = State#state{waiting_queue = queue:in({OriginPid, MsgID, Msg}, WaitingQueue)}, 5 {noreply, NewState, ?HIBERNATE_TIMEOUT}; 6 7 handle_cast({get_token, OriginPid, _MsgID, Msg}, 8 #state{max_num = MaxNum, time_interval = TimeInterval} = State) -> 9 10 %% do something operation 11 queue_handle(OriginPid, Msg), 12 13 %% send after 60s, tell the queue seed will active 14 erlang:send_after(TimeInterval, erlang:self(), {active}), 15 NewState = State#state{max_num = MaxNum - 1}, 16 {noreply, NewState, ?HIBERNATE_TIMEOUT};
如果无可用的活动因子(L2), 即将调用者进程写入到等待队列中.
如存在可用的活动因子(L8), 就返回给调用者继续执行的token, 设置定时, 并将计数器减一.
活动因子恢复:
1 handle_info({active}, #state{max_num = MaxNum, 2 time_interval = TimeInterval, 3 waiting_queue = WaitingQueue} = State) -> 4 case queue:out(WaitingQueue) of 5 {{value, {OriginPid, _MsgID, Msg}}, NewWaitingQueue} -> 6 case erlang:is_process_alive(OriginPid) of 7 true -> 8 %% do something operations 9 queue_handle(OriginPid, Msg), 10 11 erlang:send_after(TimeInterval, erlang:self(), {active}), 12 13 {noreply, State#state{max_num = MaxNum, 14 waiting_queue = NewWaitingQueue}, ?HIBERNATE_TIMEOUT}; 15 _ -> 16 {noreply, State#state{max_num = MaxNum + 1, 17 waiting_queue = NewWaitingQueue}, ?HIBERNATE_TIMEOUT} 18 end; 19 {empty, NewWaitingQueue} -> 20 {noreply, State#state{max_num = MaxNum + 1, 21 waiting_queue = NewWaitingQueue}, ?HIBERNATE_TIMEOUT} 22 end;
定时恢复活动因子后, 检查waiting queue 中, 是否有等待者.
至此, 基本上就很简单了.
总结
1, 利用活动因子来满足任意时间段
2, 检查等待队列深度可以避免不必要的等待