Redis 作为高性能的内存数据库,在缓存、消息队列、排行榜等场景中广泛应用。然而,要充分发挥 Redis 的性能潜力,需要从配置、数据结构、架构设计等多个维度进行优化。
Redis 不同数据结构的内存开销差异巨大:
# String:简单键值对 SET user:1001:name "John" # 内存占用:约 100 字节 # Hash:适合对象存储 HSET user:1001 name "John" age 30 email "john@example.com" # 内存占用:约 150 字节(比多个 String 节省 50%) # List:适合队列和栈 LPUSH queue:task "task1" LRANGE queue:task 0 -1 # Set:适合去重和集合运算 SADD tags:article:1001 "redis" "database" "cache" # ZSet:适合排行榜 ZADD rank:score 100 player1 200 player2
使用 Hash 代替多个 String:
# 不推荐:多个 String SET user:1001:name "John" SET user:1001:age 30 SET user:1001:email "john@example.com" # 总内存:约 300 字节 # 推荐:单个 Hash HMSET user:1001 name "John" age 30 email "john@example.com" # 总内存:约 150 字节,节省 50%
压缩列表优化:
# 当 Hash 或 ZSet 元素少于 512 个,且值小于 64 字节时 # Redis 自动使用 ziplist 编码,内存占用显著降低 # 查看编码方式 OBJECT encoding user:1001 # 输出:ziplist(内存优化)或 hashtable(常规)
# 不推荐:过长的键名 SET this_is_a_very_long_key_name_that_wastes_memory "value" # 推荐:简短但清晰的键名 SET usr:1001:nam "value" # 使用缩写:usr= user, nam= name, em= email
# redis.conf # RDB 快照频率调整 # 根据业务需求平衡性能和数据安全 save 900 1 # 900秒内至少1个key变化 save 300 10 # 300秒内至少10个key变化 save 60 10000 # 60秒内至少10000个key变化 # RDB 文件压缩 rdbcompression yes # RDB 文件校验 rdbchecksum yes # 后台持久化 stop-writes-on-bgsave-error yes
# AOF 持久化 appendonly yes # AOF 同步策略 # always: 每次写入都同步,最安全但最慢 # everysec: 每秒同步一次,推荐 # no: 由操作系统决定同步 appendfsync everysec # AOF 重写配置 auto-aof-rewrite-percentage 100 auto-aof-rewrite-min-size 64mb # AOF 文件增量同步 aof-use-rdb-preamble yes
# 最大客户端连接数 maxclients 10000 # TCP 监听队列长度 tcp-backlog 511 # TCP keepalive tcp-keepalive 300
# 不推荐:多次网络往返 for i in {1..1000}; do redis-cli SET key:$i value:$i done # 推荐:使用 Pipeline redis-cli --pipe <<EOF SET key:1 value:1 SET key:2 value:2 ... SET key:1000 value:1000 EOF # 或使用 MSET redis-cli MSET key:1 value:1 key:2 value:2 ... key:1000 value:1000
# 禁用 KEYS 命令(生产环境) # 使用 SCAN 代替 redis-cli --scan --pattern "user:*" --count 100 # 禁用 FLUSHDB/FLUSHALL rename-command FLUSHDB "" rename-command FLUSHALL ""
# 内存淘汰策略 # volatile-lru: 从设置了过期时间的键中驱逐 # allkeys-lru: 从所有键中驱逐(推荐) # volatile-random: 随机驱逐设置了过期时间的键 # allkeys-random: 随机驱逐所有键 # volatile-ttl: 驱逐即将过期的键 # noeviction: 不驱逐,写入时返回错误 maxmemory-policy allkeys-lru # 最大内存限制 maxmemory 2gb
# 主节点配置 port 6379 bind 0.0.0.0 # 从节点配置 port 6380 replicaof 127.0.0.1 6379 # 从节点只读 replica-read-only yes
# sentinel.conf port 26379 sentinel monitor mymaster 127.0.0.1 6379 2 sentinel down-after-milliseconds mymaster 5000 sentinel parallel-syncs mymaster 1 sentinel failover-timeout mymaster 10000
# cluster-enabled yes # cluster-config-file nodes.conf # cluster-node-timeout 5000 # cluster-require-full-coverage yes # 集群分片槽 # 16384 个槽位分布在多个节点
# 使用哈希标签确保相关数据在同一节点 # {user:1001}:profile 和 {user:1001}:posts 在同一节点 SET {user:1001}:profile "..." SET {user:1001}:posts "..."
import redis r = redis.Redis(host='localhost', port=6379, db=0) def get_user(user_id): # 先查缓存 cache_key = f"user:{user_id}" user = r.get(cache_key) if user: return user # 缓存未命中,查数据库 user = db.query(f"SELECT * FROM users WHERE id = {user_id}") # 写入缓存,设置过期时间 r.setex(cache_key, 3600, user) return user
def warm_up_cache(): # 应用启动时预加载热点数据 hot_users = db.query("SELECT * FROM users WHERE is_hot = 1") pipe = r.pipeline() for user in hot_users: pipe.setex(f"user:{user.id}", 3600, user) pipe.execute()
def update_user(user_id, data): # 先更新数据库 db.execute(f"UPDATE users SET ... WHERE id = {user_id}") # 再删除缓存(而不是更新) # 避免并发更新导致数据不一致 r.delete(f"user:{user_id}")
# 查看内存使用 redis-cli INFO memory # 查看连接数 redis-cli INFO clients # 查看命令统计 redis-cli INFO commandstats # 查看持久化状态 redis-cli INFO persistence # 查看复制状态 redis-cli INFO replication
# 配置慢查询阈值 CONFIG SET slowlog-log-slower-than 10000 # 10ms CONFIG SET slowlog-max-len 128 # 查看慢查询 SLOWLOG GET 10 # 分析慢查询 SLOWLOG GET 1 1) 1) (integer) 1234567890 2) (integer) 15000 3) "KEYS" "user:*"
# 使用 redis-benchmark 进行性能测试 redis-benchmark -h 127.0.0.1 -p 6379 -c 50 -n 100000 # 测试特定命令 redis-benchmark -t set,get,lpush,lpop -n 100000 -c 50
def update_score(user_id, score): # 使用 ZSet 实现排行榜 key = f"rank:daily:{datetime.now().strftime('%Y%m%d')}" r.zadd(key, {user_id: score}) # 设置过期时间(7天) r.expire(key, 604800) def get_top_users(limit=100): key = f"rank:daily:{datetime.now().strftime('%Y%m%d')}" # 按分数倒序获取 return r.zrevrange(key, 0, limit-1, withscores=True) def get_user_rank(user_id): key = f"rank:daily:{datetime.now().strftime('%Y%m%d')}" return r.zrevrank(key, user_id)
def increment_counter(counter_name): key = f"counter:{counter_name}:{datetime.now().strftime('%Y%m%d')}" return r.incr(key) def get_counter_stats(counter_name, days=7): pipe = r.pipeline() for i in range(days): date = (datetime.now() - timedelta(days=i)).strftime('%Y%m%d') key = f"counter:{counter_name}:{date}" pipe.get(key) results = pipe.execute() return results
import time import uuid def acquire_lock(lock_name, acquire_timeout=10, lock_timeout=10): identifier = str(uuid.uuid4()) lock_key = f"lock:{lock_name}" end_time = time.time() + acquire_timeout while time.time() < end_time: # 尝试获取锁 if r.setnx(lock_key, identifier): r.expire(lock_key, lock_timeout) return identifier # 锁已存在,检查是否过期 if r.ttl(lock_key) == -1: r.expire(lock_key, lock_timeout) time.sleep(0.001) return False def release_lock(lock_name, identifier): lock_key = f"lock:{lock_name}" # 使用 Lua 脚本确保原子性 lua_script = """ if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end """ result = r.eval(lua_script, 1, lock_key, identifier) return result == 1
# 查找大键 redis-cli --bigkeys # 查找过期键 redis-cli --scan --pattern "*:*" | xargs redis-cli TTL # 分析内存使用 redis-cli MEMORY USAGE key:name
# 查看阻塞操作 redis-cli CLIENT LIST # 查看长事务 redis-cli CLIENT LIST | grep age # 查看慢查询 redis-cli SLOWLOG GET
# 查看客户端连接 redis-cli CLIENT LIST # 杀掉特定客户端 redis-cli CLIENT KILL 127.0.0.1:12345 # 查看连接数 redis-cli INFO clients
user:1001:profile# 绑定特定 IP bind 127.0.0.1 # 设置密码 requirepass your_strong_password # 禁用危险命令 rename-command CONFIG "" rename-command SHUTDOWN "" rename-command FLUSHDB "" rename-command FLUSHALL ""
Redis 性能优化是一个系统工程:
✅ 内存优化:选择合适的数据结构,优化键命名
✅ 持久化优化:根据场景选择 RDB 或 AOF
✅ 网络优化:使用批量操作减少网络往返
✅ 架构优化:主从复制、集群模式、缓存策略
✅ 监控优化:持续监控关键指标,及时发现问题
通过合理的配置和架构设计,Redis 可以轻松支撑数十万 QPS 的业务场景。