社交网站已经成为人类生活中不可或缺的部分。现有的两大社交网站——微博，推特都高度依赖 Redis 承载海量用户访问。
本实训项目从创建用户与状态，处理用户关系，状态与信息流三个方面介绍如何使用 Redis 构建简单的社交网站。
第1关：创建用户与动态
#!/usr/bin/env python #-*- coding:utf-8 -*- import re import time import redis conn = redis.Redis() # 创建新用户 def create_user(login_name, real_name): # 请在下面完成要求的功能 #********* Begin *********# login_name = login_name.lower() if conn.hget("users", login_name): return None uid = conn.incr("user:id") pipe = conn.pipeline(True) pipe.hset("users", login_name, uid) pipe.hmset("user:%i"%(uid), { 'login_name': login_name, 'id': uid, 'real_name': real_name, 'followers': 0, 'following': 0, 'posts': 0, 'last_signup': time.time(), }) pipe.execute() return uid #********* End *********# # 为用户创建新动态 def create_post(uid, content): # 请在下面完成要求的功能 #********* Begin *********# pipe = conn.pipeline(True) pipe.hget("user:%i"%(uid), 'login_name') pipe.incr("post:id") login_name, pid = pipe.execute() if not login_name: return None pipe.hmset("post:%i"%(pid), { 'id': pid, 'uid': uid, 'content': content, 'posted': time.time(), 'user_name': login_name, }) pipe.hincrby("user:%i"%(uid), 'posts') pipe.execute() return pid #********* End *********#第2关：处理用户关系
#!/usr/bin/env python #-*- coding:utf-8 -*- import re import time import redis conn = redis.Redis() # 关注用户 def follow(uid, other_uid): # 请在下面完成要求的功能 #********* Begin *********# fkey1 = "following:%s"%(uid) fkey2 = "followers:%s"%(other_uid) if conn.zscore(fkey1, other_uid): return None now = time.time() pipe = conn.pipeline(True) pipe.zadd(fkey1, other_uid, now) pipe.zadd(fkey2, uid, now) following, followers = pipe.execute() pipe.hincrby("user:%s"%(uid), 'following', int(following)) pipe.hincrby("user:%s"%(other_uid), 'followers', int(followers)) pipe.execute() return True #********* End *********# # 取消关注 def unfollow(uid, other_uid): # 请在下面完成要求的功能 #********* Begin *********# fkey1 = "following:%s"%(uid) fkey2 = "followers:%s"%(other_uid) if not conn.zscore(fkey1, other_uid): return None pipe = conn.pipeline(True) pipe.zrem(fkey1, other_uid) pipe.zrem(fkey2, uid) following, followers = pipe.execute() pipe.hincrby("user:%s"%(uid), 'following', -int(following)) pipe.hincrby("user:%s"%(other_uid), 'followers', -int(followers)) pipe.execute() return True #********* End *********# # 创建新用户 def create_user(login_name, real_name): login_name = login_name.lower() if conn.hget("users", login_name): return None uid = conn.incr("user:id") pipe = conn.pipeline(True) pipe.hset("users", login_name, uid) pipe.hmset("user:%i"%(uid), { 'login_name': login_name, 'id': uid, 'real_name': real_name, 'followers': 0, 'following': 0, 'posts': 0, 'last_signup': time.time(), }) pipe.execute() return uid # 为用户创建新动态 def create_post(uid, content): pipe = conn.pipeline(True) pipe.hget("user:%i"%(uid), 'login_name') pipe.incr("post:id") login_name, pid = pipe.execute() if not login_name: return None pipe.hmset("post:%i"%(pid), { 'id': pid, 'uid': uid, 'content': content, 'posted': time.time(), 'user_name': login_name, }) pipe.hincrby("user:%i"%(uid), 'posts') pipe.execute() return pid第3关：状态与信息流
#!/usr/bin/env python #-*- coding:utf-8 -*- import re import time import redis conn = redis.Redis() # 获得主页时间线 def get_home_timeline(uid, page=1, count=30): # 请在下面完成要求的功能 #********* Begin *********# post_ids = conn.zrevrange("home:%s"%(uid), 0, -1) pipe = conn.pipeline(True) for pid in post_ids: pipe.hgetall("post:%s"%(pid)) return pipe.execute() #********* End *********# # 发布动态并将动态推送给粉丝 def post(uid, content): # 请在下面完成要求的功能 #********* Begin *********# pid = create_post(uid, content) if not pid: return None posted = conn.hget("post:%s"%(pid), "posted") conn.zadd("profile:%s"%(uid), pid, float(posted)) followers = conn.zrange("followers:%s"%(uid), 0, -1) pipe = conn.pipeline(False) for follower in followers: pipe.zadd("home:%s"%(follower), pid, float(posted)) pipe.execute() return pid #********* End *********# # 关注用户 def follow(uid, other_uid): fkey1 = "following:%s"%(uid) fkey2 = "followers:%s"%(other_uid) if conn.zscore(fkey1, other_uid): return None now = time.time() pipe = conn.pipeline(True) pipe.zadd(fkey1, other_uid, now) pipe.zadd(fkey2, uid, now) following, followers = pipe.execute() posts = conn.zrevrange("profile:%s"%(other_uid), 0, 100, withscores=True) if posts: pipe.zadd("home:%s"%(uid), **dict(posts)) pipe.hincrby("user:%s"%(uid), 'following', int(following)) pipe.hincrby("user:%s"%(other_uid), 'followers', int(followers)) pipe.execute() return True # 取消关注 def unfollow(uid, other_uid): fkey1 = "following:%s"%(uid) fkey2 = "followers:%s"%(other_uid) if not conn.zscore(fkey1, other_uid): return None pipe = conn.pipeline(True) pipe.zrem(fkey1, other_uid) pipe.zrem(fkey2, uid) following, followers = pipe.execute() posts = conn.zrevrange("profile:%s"%(other_uid), 0, -1) if posts: pipe.zrem("home:%s"%(uid), *posts) pipe.hincrby("user:%s"%(uid), 'following', -int(following)) pipe.hincrby("user:%s"%(other_uid), 'followers', -int(followers)) pipe.execute() return True # 创建新用户 def create_user(login_name, real_name): login_name = login_name.lower() if conn.hget("users", login_name): return None uid = conn.incr("user:id") pipe = conn.pipeline(True) pipe.hset("users", login_name, uid) pipe.hmset("user:%i"%(uid), { 'login_name': login_name, 'id': uid, 'real_name': real_name, 'followers': 0, 'following': 0, 'posts': 0, 'last_signup': time.time(), }) pipe.execute() return uid # 为用户创建新动态 def create_post(uid, content): pipe = conn.pipeline(True) pipe.hget("user:%i"%(uid), 'login_name') pipe.incr("post:id") login_name, pid = pipe.execute() if not login_name: return None pipe.hmset("post:%i"%(pid), { 'id': pid, 'uid': uid, 'content': content, 'posted': time.time(), 'user_name': login_name, }) pipe.hincrby("user:%i"%(uid), 'posts') pipe.execute() return pid