一、使用EXPIRE进行访问频率限制
逻辑：用户登入网页，判断是否为第一次登入，如果是，创建一个新键记录访问次数为1，并设置超时时间。后面再登入时先判断键是否存在且大于访问次数限制，如果是的话，返回错误页面，否则访问次数+1，正常访问。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 |
# -*- coding: utf8 -*- import web import redis """ LIMIT_TIME：在多长的时间范围内 LIMIT_TIMES：最多访问多少次 """ LIMIT_TIMES = 3 LIMIT_TIME = 60 conn = redis.StrictRedis() """ 路由处理 """ urls = ( "/", "Index" ) class Index: @staticmethod def __mk_h1(data): return "<title>HelloWorld</title><div align=center><h1>%s</h1></div>" % data def __not_first_visit(self, key_name, user): """ 第一次访问时的操作 :return: 返回网页内容 """ # 键值存在，获取数量 n = int(conn.get(key_name)) print key_name, n if n >= LIMIT_TIMES: # 如果数量超过指定限制，返回错误 return self.__mk_h1("Error") else: # 返回正常的页面 conn.incr(key_name) return self.__mk_h1("Hello %s! This is you %d visit!" % (user, n + 1)) def __first_visit(self, key_name, user): """ 非第一次访问的操作 :return: 返回网页内容 """ # 键不存在，创建新键 pipe = conn.pipeline() pipe.incr(key_name) pipe.expire(key_name, LIMIT_TIME) pipe.execute() return self.__mk_h1("Hello %s, This is you first visit!" % user) def GET(self): """ 处理用户请求 """ params = web.input() if "user" not in params: return self.__mk_h1("Who are you") else: user = params["user"] key_name = "login:times:%s" % params["user"] if conn.exists(key_name) is True: return self.__not_first_visit(key_name, user) else: return self.__first_visit(key_name, user) if __name__ == "__main__": app = web.application(urls, globals()) app.run() |
测试，在网页中输入https://127.0.0.1:8080?user=maqian，并刷新三次：
当第三次刷新也就是第四次访问时，页面会返回错误：
二、使用列表限制登陆频率
使用上面的方法有一个问题，假设用户在某一分钟的第一秒访问了一次，然后在最后一秒访问两次，此时进入第二秒，用户又立马访问两次，根据上面的规则，这两次访问是可以正常访问的，但是这样就导致用户在两秒内访问了4次，和我们的要求一分钟最多3次不符。
此时可以使用列表来完善这个功能：当用户访问次数小于三次时，把当前访问时间插入列表，让用户正常访问。大于三次时，提取出第一次访问和当前时间进行比较，如果时间间隔小于规定时间，返回错误，否则，删掉第一个记录，并插入新的纪录在最后。
修改上面Index类代码为:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 |
class Index: @staticmethod def __mk_h1(data): return "<title>HelloWorld</title><div align=center><h1>%s</h1></div>" % data def __visit(self, key_name): n = int(conn.llen(key_name)) if n < LIMIT_TIMES: conn.lpush(key_name, time.time()) return self.__mk_h1("Hello!") else: now = time.time() t = float(conn.lrange(key_name, 2, 2)[0]) # 取出最新的时间 print now, t, now - t if time.time() - t <= LIMIT_TIME: return self.__mk_h1("Error") else: # 弹出最右边的元素，即最先被插进来的元素 conn.rpop(key_name) # 插入元素 conn.lpush(key_name, now) return self.__mk_h1("Hello") def GET(self): """ 处理用户请求 """ params = web.input() if "user" not in params: return self.__mk_h1("Who are you") else: user = params["user"] key_name = "login:times:%s" % user return self.__visit(key_name) |
测试，正常访问：
一旦任何20S内访问次数超过三次，返回错误页面：
评论