python 限制单位时间内函数执行次数

Posted 2021-05-10

#https://gist.github.com/gregburek/1441055
#限制单位时间内函数执行次数
import time
from collections import deque


def RateLimited(minutes=1, limitNum=3):
    status = deque([0] * limitNum, maxlen=limitNum)
    def decorate(func):
        def wrap(*args, **kwargs):
            current = time.clock()
            elapsed = current - status[0]
            leftToWait = minutes * 60 - elapsed
            if leftToWait > 0 and status[0] != 0:
                time.sleep(abs(leftToWait))
            ret = func(*args, **kwargs)
            status.append(time.clock())
            return ret
        return wrap
    return decorate

def redis_limit(sec=1, limit_num=30):
    rd = redis.Redis(db=1, decode_responses=True)
    name = "redis_limit"
    rd.delete(name)

    def decorate(func):
        def wrap(*args, **kwargs):
            current = time.time()
            first_status = int(float(rd.lindex(name, -1) or 0))
            elapsed = current - first_status
            wait = sec - elapsed
            if wait > 0 and rd.llen(name) >= limit_num:
                time.sleep(abs(wait))
            ret = func(*args, **kwargs)
            rd.lpush(name, current)
            rd.ltrim(name, start=0, end=limit_num - 1)
            return ret

        return wrap

    return decorate

@RateLimited()
def dd(i):
    time.sleep(3)
    print i

for i in range(200):
    dd(i)