这个练习来自 《Redis 实战》,但有所简化,主要练习一下redis 的使用。主要是构建一个文章点赞网站,用户可以在网站上发布文章和对其他用户的文章点赞,然后网站可根据文章的发布时间展示最近发布的文章,和根据点赞数量展示拥有最多点赞数的文章。
redis 客户端
import redis
redis_client = redis.Redis(host='192.168.64.2', port=30379, db=0)
复制代码
发布文章
发布文章的代码如下,我们用到了
- incr
- sadd
- hmset
- zadd
这几个命令。其中:
incr 命令将 key 中储存的数字值增一。如果 key 不存在,那么 key 的值会先被初始化为 0 ,然后再执行 incr 操作。我们使用它作为文章的key。
sadd 命令将一个或多个成员元素加入到集合中,已经存在于集合的成员元素将被忽略。假如集合 key 不存在,则创建一个只包含添加的元素作成员的集合。
我们使用它将作者加入到点赞这篇文章的用户列表中。
hmset 命令用于同时将多个 field-value (字段-值)对设置到哈希表中。此命令会覆盖哈希表中已存在的字段。我们使用它来缓存文章信息。
zadd 命令用于将一个或多个成员元素及其分数值加入到有序集当中。如果某个成员已经是有序集的成员,那么更新这个成员的分数值,并通过重新插入这个成员元素,来保证该成员在正确的位置上。我们使用它来做文章的排序。
def post_article(user):
"""
发表一篇文章
"""
# 利用redis的计数器生成文章id
article_id = str(redis_client.incr('article:'))
# 文章title
article_title = f'article_title_{article_id}'
# 文章发布时间
now = time.time()
# 将作者加入到点赞这篇文章的用户列表中
voted = f'voted:{article_id}'
redis_client.sadd(voted, user)
# 将文章信息存放到一个散列中
article = f'article:{article_id}'
redis_client.hmset(
article,
{
'title': article_title,
'poster': user,
'time': now,
'votes': 1,
},
)
# 将文章发布到根据点赞数量排序的有序集合中
redis_client.zadd('score:', {article: 1})
# 将文章发布到根据发布时间排序的有序集合中
redis_client.zadd('time:', {article: now})
return article_id
复制代码
点赞文章
点赞文章的代码如下,我们用到了
- zincrby
- hincrby
这两个新命令。其中:
zincrby 命令对有序集合中指定成员的分数加上增量 ,也可以通过传递一个负数值,让分数减去相应的值。
我们用它来改变文章点赞的有序集和里文章的点赞数目。
hincrby 命令用于为哈希表中的字段值加上指定增量值。增量也可以为负数,相当于对指定字段进行减法操作。我们用它来改变文章详细信息里的点赞数目。
def vote_article(user, article):
"""
为文章点赞
"""
article_id = article.split(':')[-1]
if redis_client.sadd('voted:' + article_id, user):
redis_client.zincrby('score:', 1, article)
redis_client.hincrby(article, 'votes', 1)
复制代码
获取文章
获取文章的代码如下,我们用到了
- zrevrange
这个新命令。
zrevrange 命令返回有序集中,指定区间内的成员。其中成员的位置按分数值递减(从大到小)来排列。我们用它来获取被点赞数最多的10篇文章,或者最近发表的10篇文章。
def get_articles(order='score'):
"""
获取 10 篇文章
"""
ids = redis_client.zrevrange(order + ':', 0, 10)
articles = []
for id in ids:
article_data = redis_client.hgetall(id)
article_data['id'] = id
articles.append(article_data)
return articles
复制代码
使用 FastAPI 构建后端
@app.get('/')
def read_root():
return {'Welcome to learn Redis'}
@app.get('/article')
def get_article(order: Optional[str] = Query('score', enum=['score', 'time'])):
return get_articles(order)
@app.post('/random_post_article')
def random_post_article(number: Optional[int] = 10):
"""
使用随机用户发表文章
"""
for _ in range(number):
user = randrange(number)
post_article(user)
return True
@app.post('/random_vote')
def random_vote(number: Optional[int] = 10):
"""
使用随机用户对随机文章点赞
"""
article_id_max = int(redis_client.get('article:'))
for _ in range(number):
user = randrange(number)
article = f'article:{randrange(1, article_id_max+1)}'
vote_article(user, article)
return True
@app.post('/clear_data')
def clear_data():
"""
清空数据
"""
for key in redis_client.scan_iter('*'):
redis_client.delete(key)
复制代码
启动服务后访问 http://localhost:8000/docs 即可进行测试。
近期评论