1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62
| class LeaderboardService: def __init__(self, redis_client): self.redis = redis_client def update_score(self, leaderboard_name, user_id, score): """更新用户分数""" key = f"leaderboard:{leaderboard_name}" self.redis.zadd(key, {user_id: score}) self.redis.expire(key, 86400) def get_rank(self, leaderboard_name, user_id): """获取用户排名""" key = f"leaderboard:{leaderboard_name}" rank = self.redis.zrevrank(key, user_id) score = self.redis.zscore(key, user_id) if rank is not None: return { "rank": rank + 1, "score": score } return None def get_top_players(self, leaderboard_name, count=10): """获取排行榜前N名""" key = f"leaderboard:{leaderboard_name}" players = self.redis.zrevrange(key, 0, count - 1, withscores=True) return [ {"user_id": user_id, "score": score} for user_id, score in players ] def get_around_user(self, leaderboard_name, user_id, radius=2): """获取用户附近的排名""" key = f"leaderboard:{leaderboard_name}" rank = self.redis.zrevrank(key, user_id) if rank is None: return [] start = max(0, rank - radius) end = rank + radius players = self.redis.zrevrange(key, start, end, withscores=True) return [ {"user_id": user_id, "score": score, "rank": start + i + 1} for i, (user_id, score) in enumerate(players) ]
redis_client = redis.Redis(host='localhost', port=6379, db=0) leaderboard = LeaderboardService(redis_client)
leaderboard.update_score("game_ranking", "player123", 1500)
rank_info = leaderboard.get_rank("game_ranking", "player123") top_players = leaderboard.get_top_players("game_ranking", 5)
|