Python Redis字符串案例如何操作数据

wen python案例 2026-06-30 30

本文目录导读：

基本环境配置
字符串基本操作
高级字符串操作
数值操作
字符串范围操作
实际应用案例
常用辅助操作
性能优化技巧

我来为你详细介绍 Python 操作 Redis 字符串的核心案例和操作数据的方法。

基本环境配置

安装依赖

pip install redis

连接 Redis

import redis
# 连接 Redis
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
# 连接池方式（推荐）
pool = redis.ConnectionPool(host='localhost', port=6379, db=0)
r = redis.Redis(connection_pool=pool, decode_responses=True)

字符串基本操作

设置和获取值

# 设置单个键值对
r.set('name', '张三')
r.set('age', 25)
# 获取值
name = r.get('name')  # 返回: '张三'
print(name)
# 获取不存在的键
result = r.get('not_exist')  # 返回: None
print(result)

批量操作

# 批量设置
r.mset({
    'user:1:name': '张三',
    'user:1:age': '25',
    'user:1:city': '北京'
})
# 批量获取
values = r.mget(['user:1:name', 'user:1:age', 'user:1:city'])
print(values)  # ['张三', '25', '北京']

高级字符串操作

设置带过期时间的值

# 方式1: 使用 setex
r.setex('temp_code', 60, '123456')  # 60秒过期
# 方式2: 使用 set 的 ex 参数
r.set('verification_code', '789012', ex=300)  # 300秒过期
# 方式3: 使用 expire
r.set('session_token', 'token_123')
r.expire('session_token', 3600)  # 1小时后过期
# 查看剩余过期时间
ttl = r.ttl('temp_code')
print(f'剩余时间: {ttl}秒')

只有当键不存在时才设置

# setnx (SET if Not eXists)
result = r.setnx('unique_key', 'value')
if result:
    print('设置成功，键之前不存在')
else:
    print('键已存在，设置失败')
# 带过期时间的 setnx
r.set('lock:user:1', '1', nx=True, ex=10)  # 分布式锁实现

数值操作

# 设置数值
r.set('counter', 0)
r.set('score', 100)
# 递增操作
r.incr('counter')  # 加1，结果: 1
r.incr('counter', 5)  # 加5，结果: 6
r.incrby('counter', 10)  # 加10，结果: 16
# 递减操作
r.decr('score')  # 减1，结果: 99
r.decr('score', 20)  # 减20，结果: 79
r.decrby('score', 5)  # 减5，结果: 74
# 浮点数操作
r.set('price', '10.5')
r.incrbyfloat('price', 3.2)  # 结果: 13.7
# 获取旧值并设置新值
old_value = r.getset('counter', '100')
print(f'旧值: {old_value}, 新值: {r.get("counter")}')

字符串范围操作

# 设置字符串
r.set('message', 'Hello, Redis!')
# 获取子字符串
sub = r.getrange('message', 0, 4)  # 返回: 'Hello'
print(sub)
# 设置子字符串
r.setrange('message', 7, 'World')  # 结果: 'Hello, World!'
print(r.get('message'))
# 获取字符串长度
length = r.strlen('message')  # 返回: 13
print(f'字符串长度: {length}')

实际应用案例

案例：用户登录限制

import time
class LoginLimiter:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.max_attempts = 5
        self.lock_time = 300  # 5分钟
    def check_login(self, username):
        """检查登录限制"""
        key = f"login:limit:{username}"
        attempts = self.redis.get(key)
        if attempts and int(attempts) >= self.max_attempts:
            ttl = self.redis.ttl(key)
            return False, f"账户已锁定，请{ttl}秒后重试"
        return True, "可以登录"
    def record_attempt(self, username):
        """记录登录尝试"""
        key = f"login:limit:{username}"
        # 递增尝试次数
        attempts = self.redis.incr(key)
        # 第一次尝试时设置过期时间
        if attempts == 1:
            self.redis.expire(key, self.lock_time)
        remaining = self.max_attempts - attempts
        return remaining
# 使用示例
limiter = LoginLimiter(r)
# 模拟登录
username = "user123"
can_login, msg = limiter.check_login(username)
print(f"状态: {msg}")
if can_login:
    # 模拟登录失败
    remaining = limiter.record_attempt(username)
    print(f"剩余尝试次数: {remaining}")

案例：缓存管理

import json
class CacheManager:
    def __init__(self, redis_client, default_ttl=3600):
        self.redis = redis_client
        self.default_ttl = default_ttl
    def get(self, key):
        """获取缓存"""
        data = self.redis.get(key)
        if data:
            return json.loads(data)
        return None
    def set(self, key, value, ttl=None):
        """设置缓存"""
        ttl = ttl or self.default_ttl
        self.redis.setex(key, ttl, json.dumps(value, ensure_ascii=False))
    def delete(self, key):
        """删除缓存"""
        self.redis.delete(key)
    def exists(self, key):
        """检查缓存是否存在"""
        return self.redis.exists(key)
    def clear_pattern(self, pattern):
        """清除匹配模式的缓存"""
        keys = self.redis.keys(pattern)
        if keys:
            self.redis.delete(*keys)
# 使用示例
cache = CacheManager(r, default_ttl=7200)
# 缓存用户数据
user_data = {
    "id": 1,
    "name": "张三",
    "email": "zhangsan@example.com"
}
cache.set("user:1:data", user_data, ttl=3600)
# 获取缓存数据
cached_user = cache.get("user:1:data")
print(cached_user["name"])  # 张三

常用辅助操作

# 检查键是否存在
exists = r.exists('name')
print(f'键存在: {exists}')
# 查找键
keys = r.keys('user:*')  # 查找所有以user:开头的键
print(f'匹配的键: {keys}')
# 获取键类型
key_type = r.type('name')  # 返回: 'string'
print(f'键类型: {key_type}')
# 删除键
r.delete('temp_key')
# 检查并设置（CAS操作）
with r.pipeline() as pipe:
    while True:
        try:
            pipe.watch('counter')
            current = int(pipe.get('counter') or 0)
            pipe.multi()
            pipe.set('counter', current + 1)
            pipe.execute()
            break
        except redis.WatchError:
            continue

性能优化技巧

# 使用 pipeline 批量操作
pipe = r.pipeline()
for i in range(1000):
    pipe.set(f'key:{i}', f'value:{i}')
pipe.execute()  # 一次网络往返完成所有操作
# 使用 eval 执行 Lua 脚本
lua_script = """
local key = KEYS[1]
local value = ARGV[1]
local ttl = tonumber(ARGV[2])
local current = redis.call('GET', key)
if not current then
    redis.call('SETEX', key, ttl, value)
    return 1
end
return 0
"""
set_if_not_exists = r.register_script(lua_script)
result = set_if_not_exists(keys=['unique_key'], args=['value', '3600'])

这些代码示例涵盖了 Python 操作 Redis 字符串的核心方法，你可以根据实际需求灵活运用，记住要根据业务场景选择合适的操作方式和数据类型。