网站中经常需要用到网站限流，例如限制用户恶意爬虫获取网页信息以及在网站访问流量大时限制访问频率等等，这里使用redis做一个简单的网站登录限制案例。
一、使用EXPIRE进行访问频率限制
逻辑：用户登入网页，判断是否为第一次登入，如果是，创建一个新键记录访问次数为1，并设置超时时间。后面再登入时先判断键是否存在且大于访问次数限制，如果是的话，返回错误，否则就把访问次数+1，返回正常页面。
# -*- 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()
测试，在网页中输入http://127.0.0.1:8080?user=maqian，并刷新三次：
当第三次刷新也就是第四次访问时，页面会返回错误：
二、使用列表限制登陆频率
使用上面的方法有一个问题，假设用户在某一分钟的第一秒访问了一次，然后在最后一秒访问两次，此时进入第二秒，用户又立马访问两次，根据上面的规则，这两次访问是可以正常访问的，但是这样就导致用户在两秒内访问了4次，和我们的要求不符。
此时可以使用列表来完善这个功能：当用户访问次数小于三次时，把当前访问时间插入列表，让用户正常访问。大于三次时，提取出第一次访问和当前时间进行比较，如果时间间隔小于规定时间，返回错误，否则，删掉第一个记录，并插入新的纪录在最后。
修改上面Index类代码为:
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内访问次数超过三次，返回错误页面：
