网站限流经常用在网站上，比如限制用户恶意爬虫获取网页信息，在网站访问流量大时限制访问频率等。这里以redis作为一个简单的网站登录限制案例.
原文链接：Redis小案例（一）：实现网站访问频率限制
一、使用 EXPIRE 限制访问频率
逻辑：用户登录网页，判断是否为第一次登录。如果是，则新建一个key，记录访问次数为1，并设置超时时间。稍后登录时，先判断key是否存在且大于访问次数，如果存在则返回错误，否则访问次数加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 "HelloWorld %s
" % 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()
测试网站浏览有次数限制怎么办，在网页中输入：8080?user=maqian，刷新3次：
当第三次刷新，即第四次访问时，页面会返回错误：
二、使用列表限制登录频率
使用上面的方法有问题。假设用户在某分钟的第一秒访问了一次，然后在最后一秒访问了两次。此时进入第二秒，用户立即访问两次。按照上面的规则，这两次访问是可以正常访问的，但是这样会导致用户在两秒内访问了4次，不符合我们的要求。
这时候可以用一个列表来改进这个功能：当用户访问次数小于3次时，将当前访问时间插入到列表中，让用户可以正常访问。超过3次时，提取第一次访问并比较当前时间，如果时间间隔小于指定时间，则返回错误，否则网站浏览有次数限制怎么办，删除第一条记录，并在末尾插入一条新记录。
将上述Index类代码修改为：
class Index:
@staticmethod
def __mk_h1(data):
return "HelloWorld %s
" % 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内访问次数超过3次，返回错误页面：
