Python自动化操作Redis的15个实用脚本Redis作为高性能的键值数据库在现代应用开发中扮演着重要角色。Python通过redis-py库提供了丰富的Redis操作接口能够帮助开发者高效地实现数据缓存、会话管理、消息队列等多种功能。本文将介绍15个实用的自动化脚本涵盖数据操作、性能优化和高级功能等多个典型场景。环境准备首先确保已安装redis-py库pip install redis所有示例默认连接本地Redis服务器localhost:6379数据库0实际使用时请根据生产环境调整连接参数。1. 基础键值操作场景简单的数据存储与检索import redis def basic_key_operations(): r redis.Redis(hostlocalhost, port6379, db0) # 设置带过期时间的键值 r.setex(session:user123, 3600, active) # 获取值 status r.get(session:user123) print(fSession status: {status.decode() if status else Not found}) # 检查键是否存在 exists r.exists(session:user123) print(fKey exists: {exists}) if __name__ __main__: basic_key_operations()2. 批量操作键值场景高效处理大量数据def batch_operations(): r redis.Redis(hostlocalhost, port6379, db0) # 批量设置多个键值 data { user:1:name: Alice, user:1:email: aliceexample.com, user:1:age: 25 } r.mset(data) # 批量获取多个键值 keys [user:1:name, user:1:email, user:1:age] values r.mget(keys) for key, value in zip(keys, values): print(f{key}: {value.decode() if value else None})3. 哈希表高级操作场景存储结构化对象def advanced_hash_operations(): r redis.Redis(hostlocalhost, port6379, db0) user_id user:1001 # 设置多个字段 user_data { name: Bob Smith, email: bobexample.com, score: 95, last_login: 2025-12-25 } r.hset(user_id, mappinguser_data) # 获取所有字段 all_fields r.hgetall(user_id) print(User data:) for field, value in all_fields.items(): print(f {field.decode()}: {value.decode()}) # 增加数值字段 r.hincrby(user_id, score, 5) new_score r.hget(user_id, score) print(fUpdated score: {new_score.decode()})4. 列表队列操作场景实现任务队列def queue_operations(): r redis.Redis(hostlocalhost, port6379, db0) queue_key task_queue # 生产者添加任务到队列 tasks [process_data, send_email, generate_report, cleanup] for task in tasks: r.lpush(queue_key, task) print(fAdded task: {task}) # 消费者处理任务 while True: task r.brpop(queue_key, timeout1) if task: task_name task[1].decode() print(fProcessing task: {task_name}) # 模拟任务处理 import time time.sleep(0.5) print(fCompleted: {task_name}) else: print(No more tasks in queue) break5. 集合运算场景用户标签系统def set_operations(): r redis.Redis(hostlocalhost, port6379, db0) # 用户标签集合 r.sadd(user:101:tags, python, redis, backend) r.sadd(user:102:tags, python, frontend, javascript) r.sadd(user:103:tags, redis, database, backend) # 查找共同标签 common_tags r.sinter(user:101:tags, user:103:tags) print(Common tags between user 101 and 103:) for tag in common_tags: print(f - {tag.decode()}) # 合并所有标签 all_tags r.sunion(user:101:tags, user:102:tags, user:103:tags) print(\nAll unique tags:) for tag in sorted(all_tags): print(f - {tag.decode()})6. 有序集合排行榜场景游戏得分排行榜def leaderboard_system(): r redis.Redis(hostlocalhost, port6379, db0) leaderboard_key game:leaderboard # 添加玩家得分 players { player1: 1500, player2: 2200, player3: 1800, player4: 2500, player5: 1900 } r.zadd(leaderboard_key, players) # 获取前3名 top_players r.zrevrange(leaderboard_key, 0, 2, withscoresTrue) print(Top 3 Players:) for rank, (player, score) in enumerate(top_players, 1): print(f{rank}. {player.decode()}: {int(score)} points) # 获取玩家排名 player_rank r.zrevrank(leaderboard_key, player3) player_score r.zscore(leaderboard_key, player3) print(f\nPlayer3 rank: {player_rank 1 if player_rank is not None else N/A}) print(fPlayer3 score: {player_score})7. 发布订阅模式场景实时通知系统import threading import time def pubsub_system(): r redis.Redis(hostlocalhost, port6379, db0) def subscriber(channel_name): pubsub r.pubsub() pubsub.subscribe(channel_name) print(fSubscriber started on channel: {channel_name}) for message in pubsub.listen(): if message[type] message: data message[data].decode() print(fReceived: {data}) if data exit: break # 启动订阅者线程 channel notifications sub_thread threading.Thread(targetsubscriber, args(channel,)) sub_thread.daemon True sub_thread.start() time.sleep(1) # 等待订阅者准备 # 发布消息 messages [ System started, User logged in, Data updated, exit ] for msg in messages: r.publish(channel, msg) time.sleep(0.5) sub_thread.join(timeout2)8. 管道批量操作场景减少网络延迟def pipeline_performance(): r redis.Redis(hostlocalhost, port6379, db0) # 不使用管道 start time.time() for i in range(100): r.set(ftemp:{i}, fvalue:{i}) without_pipeline time.time() - start # 清理数据 for i in range(100): r.delete(ftemp:{i}) # 使用管道 start time.time() pipe r.pipeline() for i in range(100): pipe.set(ftemp:{i}, fvalue:{i}) pipe.execute() with_pipeline time.time() - start print(fWithout pipeline: {without_pipeline:.3f} seconds) print(fWith pipeline: {with_pipeline:.3f} seconds) print(fPerformance improvement: {without_pipeline/with_pipeline:.1f}x)9. 事务保证原子性场景账户余额转账def transaction_example(): r redis.Redis(hostlocalhost, port6379, db0) # 初始化账户余额 r.set(account:A, 1000) r.set(account:B, 500) def transfer_funds(from_account, to_account, amount): pipe r.pipeline(transactionTrue) try: # 监视账户变化 pipe.watch(from_account, to_account) # 检查余额 from_balance int(pipe.get(from_account) or 0) if from_balance amount: print(Insufficient funds) return False # 开始事务 pipe.multi() pipe.decrby(from_account, amount) pipe.incrby(to_account, amount) pipe.execute() print(fTransferred {amount} from {from_account} to {to_account}) return True except redis.WatchError: print(Transaction failed due to concurrent modification) return False finally: pipe.reset() # 执行转账 transfer_funds(account:A, account:B, 200) # 验证结果 print(fAccount A: {r.get(account:A).decode()}) print(fAccount B: {r.get(account:B).decode()})10. Lua脚本执行场景复杂原子操作def lua_scripting(): r redis.Redis(hostlocalhost, port6379, db0) # Lua脚本原子性地检查并设置值 lua_script local key KEYS local value ARGV local ttl ARGV local current redis.call(GET, key) if current then return {false, current} else redis.call(SET, key, value) redis.call(EXPIRE, key, ttl) return {true, value} end # 注册脚本 script r.register_script(lua_script) # 执行脚本 key cache:item result script(keys[key], args[cached_data, 60]) if result[0]: print(fCache set successfully: {result[1]}) else: print(fCache already exists: {result[1]})11. 键过期与TTL管理场景缓存失效管理def expiration_management(): r redis.Redis(hostlocalhost, port6379, db0) # 设置带过期时间的键 cache_key api:response:user:123 r.setex(cache_key, 300, {name: John, status: active}) # 5分钟过期 # 检查剩余时间 ttl r.ttl(cache_key) print(fCache will expire in {ttl} seconds) # 更新过期时间 if ttl 60: # 如果剩余时间小于1分钟 r.expire(cache_key, 600) # 延长到10分钟 print(Cache expiration extended to 10 minutes) # 设置永久键 r.set(config:site_name, MyApp) r.persist(config:site_name) # 移除过期时间12. 扫描大键集合场景安全遍历大量键def scan_keys(): r redis.Redis(hostlocalhost, port6379, db0) # 使用SCAN代替KEYS避免阻塞 pattern user:* cursor 0 user_keys [] while cursor ! 0: cursor, keys r.scan(cursorcursor, matchpattern, count100) user_keys.extend(keys) print(fFound {len(user_keys)} user keys) # 批量获取用户信息 for i in range(0, len(user_keys), 10): batch user_keys[i:i10] values r.mget(batch) for key, value in zip(batch, values): if value: print(f{key.decode()}: {value.decode()[:50]}...)13. 位图操作场景用户在线状态追踪def bitmap_operations(): r redis.Redis(hostlocalhost, port6379, db0) # 记录用户每日登录状态 bitmap_key online:2025-12-25 # 用户ID 1001-1010在25日登录 for user_id in range(1001, 1011): r.setbit(bitmap_key, user_id, 1) # 检查特定用户是否登录 user_1005_status r.getbit(bitmap_key, 1005) print(fUser 1005 online: {Yes if user_1005_status else No}) # 统计在线用户数 online_count r.bitcount(bitmap_key) print(fTotal online users: {online_count}) # 查找第一个在线的用户 first_online r.bitpos(bitmap_key, 1) print(fFirst online user ID: {first_online})14. 地理空间索引场景附近位置搜索def geospatial_operations(): r redis.Redis(hostlocalhost, port6379, db0) # 添加地理位置 locations_key cities:locations cities { Beijing: (116.4074, 39.9042), Shanghai: (121.4737, 31.2304), Guangzhou: (113.2644, 23.1291), Shenzhen: (114.0579, 22.5431) } r.geoadd(locations_key, cities) # 查找附近的城市 center (116.4074, 39.9042) # 北京坐标 nearby r.georadius( locations_key, center[0], center[1], 500, # 500公里半径 unitkm, withdistTrue, sortASC ) print(Cities within 500km of Beijing:) for city, distance in nearby: print(f {city.decode()}: {distance:.1f} km) # 计算两个城市距离 distance r.geodist(locations_key, Beijing, Shanghai, unitkm) print(f\nDistance Beijing-Shanghai: {distance} km)15. 连接池与连接管理场景生产环境连接优化import redis from redis.connection import ConnectionPool def connection_pool_management(): # 创建连接池 pool ConnectionPool( hostlocalhost, port6379, db0, max_connections10, socket_timeout5, retry_on_timeoutTrue ) # 使用连接池 r redis.Redis(connection_poolpool) try: # 测试连接 r.ping() print(Redis connection successful) # 性能测试 import time start time.time() for i in range(1000): r.set(ftest:{i}, fdata:{i}) elapsed time.time() - start print(f1000 operations completed in {elapsed:.2f} seconds) print(fOperations per second: {1000/elapsed:.0f}) except redis.ConnectionError as e: print(fConnection failed: {e}) finally: # 清理连接池 pool.disconnect() print(Connection pool cleaned up)这些脚本覆盖了Redis主要应用场景从基础操作到高级功能可以帮助开发者高效地实现各种数据存储和处理需求。在实际应用中建议根据具体业务场景选择合适的操作方式并结合性能监控和优化策略可以显著提升应用性能和开发效率。“无他惟手熟尔”有需要的用起来关注「Nicholas与Pypi」获取更多Python实战