一日一技:带过期时间的缓存、全文搜索、频率限制怎么做?

语言: CN / TW / HK

摄影:产品经理

毛肚

在以前的文章里面,我给大家介绍了使用Python自带的LRU缓存实现带有过期时间的缓存: 一日一技:实现有过期时间的LRU缓存 。也讲过倒排索引: 使用倒排索引极速提高字符串搜索效率 。但这些代码对初学者来说比较难,写起来可能会出错。

实际上,这些功能其实都可以使用Redis来实现,而且每个功能只需要1分钟就能做出来。全文搜索功能在搜索英文的时候,甚至可以智能识别拼写错误的问题。

要实现这些功能,只需要做两件事:

  1. 安装Redis

  2. Python安装第三方库: walrus

安装完成以后,我们来看看它有多简单:

带过期时间的缓存装饰器

我们想实现一个装饰器,它装饰一个函数。让我在1分钟内多次访问函数的时候,使用缓存的数据;超过1分钟以后才重新执行函数的内部代码:

import time
import datetime
from walrus import Database

db = Database()
cache = db.cache()

@cache.cached(timeout=60)
def test():
    print('函数真正运行起来')
    now = datetime.datetime.now()
    return now

now = test()
print('函数返回的数据是:', now)
time.sleep(10) # 等待10秒,此时会使用缓存
print('函数返回的数据是:', test())
time.sleep(5) # 等待5秒,此时依然使用缓存
print('函数返回的数据是:', test())

time.sleep(50)  # 让时间超过缓存的时间
print('函数返回的数据是:', test())

运行效果如下图所示:

全文搜索

我们再来看看全文搜索功能,实现起来也很简单:

from walrus import Database

db = Database()
search = db.Index('xxx')  # 这个名字随便取
poem1 = 'Early in the day it was whispered that we should sail in a boat, only thou and I, and never a soul in the world would know of this our pilgrimage to no country and to no end.'
poem2 = 'Had I the heavens’ embroidered cloths,Enwrought with golden and silver light'
poem3 = 'to be or not to be, that is a question.'

search.add('docid1', poem1) # 第一个参数不能重复
search.add('docid2', poem2)
search.add('docid3', poem3)


for doc in search.search('end'):
    print(doc['content'])

运行效果如下图所示:

如果你想让他兼容拼写错误,那么可以把 search = db.Index('xxx') 改成 search = db.Index('xxx’, metaphone=True) ,运行效果如下图所示:

不过遗憾的是,这个全文搜索功能只支持英文。

频率限制

我们有时候要限制调用某个函数的频率,或者网站的某个接口要限制IP的访问频率。这个时候,使用 walrus 也可以轻松实现:

import time
from walrus import Database

db = Database()
rate = db.rate_limit('xxx', limit=5, per=60) # 每分钟只能调用5次

for _ in range(35):
    if rate.limit('xxx'):
        print('访问频率太高!')
    else:
        print('还没有触发访问频率限制')
    time.sleep(2)

运行效果如下图所示:

其中参数 limit 表示能出现多少次, per 表示在多长时间内。

rate.limit 只要传入相同的参数,那么就会开始检查这个参数在设定的时间内出现的频率。

你可能觉得这个例子并不能说明什么问题,那么我们跟FastAPI结合一下,用来限制IP访问接口的频率。编写如下代码:

from walrus import Database, RateLimitException
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

db = Database()
rate = db.rate_limit('xxx', limit=5, per=60) # 每分钟只能调用5次

app = FastAPI()


@app.exception_handler(RateLimitException)
def parse_rate_litmit_exception(request: Request, exc: RateLimitException):
    msg = {'success': False, 'msg': f'请喝杯茶,休息一下,你的ip: {request.client.host}访问太快了!'}
    return JSONResponse(status_code=429, content=msg)

@app.get('/')
def index():
    return {'success': True}


@app.get('/important_api')
@rate.rate_limited(lambda request: request.client.host)
def query_important_data(request: Request):
    data = '重要数据'
    return {'success': True, 'data': data}

上面代码定义了一个全局的异常拦截器:

@app.exception_handler(RateLimitException)
def parse_rate_litmit_exception(request: Request, exc: RateLimitException):
    msg = {'success': False, 'msg': f'请喝杯茶,休息一下,你的ip: {request.client.host}访问太快了!'}
    return JSONResponse(status_code=429, content=msg)

在整个代码的任何地方抛出了 RateLimitException 异常,就会进入这里的逻辑中。

使用装饰器 @rate.rate_limited 装饰一个路由函数,并且这个装饰器要更靠近函数。路由函数接收什么参数,它就接收什么参数。在上面的例子中,我们只接收了 request 参数,用于获取访问者的IP。发现这个IP的访问频率超过了限制,就抛出一个 RateLimitException 。于是前面定义好的全局拦截器就会拦截 RateLimitException 异常,拦截到以后返回我们定义好的报错信息。

在频率范围内访问页面,返回正常的JSON数据:

频率超过设定的值以后,访问页面就会报错,如下图所示:

walrusredis-py 进行了很好的二次封装,用起来非常顺手。除了上面我提到的三个功能外,它还可以实现几行代码生成布隆过滤器,实现自动补全功能,实现简易图数据库等等。大家可以访问它的 官方文档了解详细使用说明 [1]

参考文献

[1] 官方文档了解详细使用说明: https://walrus.readthedocs.io/en/latest/getting-started.html

未闻 Code·知识星球开放啦!

一对一答疑爬虫相关问题

职业生涯咨询

面试经验分享

每周直播分享

......

未闻 Code·知识星球期待与你相见~

一二线大厂在职员工

十多年码龄的编程老鸟

国内外高校在读学生

中小学刚刚入门的新人

“未闻 Code技术交流群” 等你来!

入群方式:添加微信“mekingname”,备注“粉丝群”(谢绝广告党,非诚勿扰!)