Python Redis过期案例如何设置数据过期

wen python案例 2026-06-30 192

本文目录导读：

安装redis-py
基础连接
设置过期时间的几种方法
完整示例代码
重要注意事项
最佳实践建议

在Redis中设置数据过期是一个很常见的需求，下面我通过Python的redis-py库来演示几种不同的过期设置方式。

安装redis-py

pip install redis

基础连接

import redis
# 连接Redis
r = redis.Redis(host='localhost', port=6379, db=0)
# 或者使用连接池
pool = redis.ConnectionPool(host='localhost', port=6379, db=0)
r = redis.Redis(connection_pool=pool)

设置过期时间的几种方法

set时直接指定过期时间

# 设置key并指定过期时间为10秒
r.set('mykey', 'hello', ex=10)
# 设置key并指定过期时间为指定的时间点（UNIX时间戳）
import datetime
expire_at = datetime.datetime.now() + datetime.timedelta(minutes=5)
r.set('mykey', 'hello', exat=int(expire_at.timestamp()))
# 设置key并指定过期时间为毫秒
r.set('mykey', 'hello', px=10000)  # 10000毫秒 = 10秒
# 设置key并指定过期时间为毫秒级别的时间戳
r.set('mykey', 'hello', pxat=int(datetime.datetime.now().timestamp() * 1000) + 10000)

设置后单独设置过期时间

# 先设置值
r.set('mykey', 'hello')
# 然后设置过期时间（秒）
r.expire('mykey', 10)  # 10秒后过期
# 或者设置过期时间（毫秒）
r.pexpire('mykey', 10000)  # 10000毫秒后过期
# 设置指定时间过期（UNIX时间戳）
import time
r.expireat('mykey', int(time.time()) + 10)  # 10秒后过期
# 毫秒级别的时间戳
r.pexpireat('mykey', int(time.time() * 1000) + 10000)

使用setex命令

# setex: 设置值并指定过期时间（秒）
r.setex('mykey', 10, 'hello')  # 10秒后过期
# psetex: 设置值并指定过期时间（毫秒）
r.psetex('mykey', 10000, 'hello')  # 10000毫秒后过期

完整示例代码

import redis
import time
from datetime import datetime, timedelta
# 连接Redis
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
def demo_basic_expire():
    """基础过期示例"""
    print("=== 基础过期示例 ===")
    # 方法1: set时设置过期时间
    r.set('user:1', 'Alice', ex=5)  # 5秒后过期
    print(f"设置user:1=Alice，5秒后过期")
    # 立即获取
    print(f"立即获取: {r.get('user:1')}")
    # 等待3秒后获取
    time.sleep(3)
    print(f"3秒后获取: {r.get('user:1')}")
    # 再等待3秒后获取（总共6秒，已过期）
    time.sleep(3)
    print(f"6秒后获取: {r.get('user:1')}")
def demo_expire_with_update():
    """更新过期时间示例"""
    print("\n=== 更新过期时间示例 ===")
    r.set('session:abc123', 'user_data')
    r.expire('session:abc123', 30)  # 初始设置30秒过期
    # 查看剩余过期时间（秒）
    ttl = r.ttl('session:abc123')
    print(f"初始过期时间: {ttl}秒")
    # 更新为10秒后过期
    r.expire('session:abc123', 10)
    ttl = r.ttl('session:abc123')
    print(f"更新后过期时间: {ttl}秒")
def demo_remove_expire():
    """移除过期时间示例"""
    print("\n=== 移除过期时间示例 ===")
    r.set('temp:data', 'temporary', ex=60)
    print(f"设置过期时间后的TTL: {r.ttl('temp:data')}")
    # 移除过期时间
    r.persist('temp:data')
    print(f"移除过期时间后的TTL: {r.ttl('temp:data')}")  # -1表示永不过期
def demo_check_expire_status():
    """检查过期状态示例"""
    print("\n=== 检查过期状态示例 ===")
    r.set('status:test', 'check')
    # TTL: 返回剩余过期时间（秒），-1表示永不过期，-2表示key不存在
    print(f"没有设置过期时间的TTL: {r.ttl('status:test')}")
    # PTTL: 返回剩余过期时间（毫秒）
    print(f"没有设置过期时间的PTTL: {r.pttl('status:test')}")
    # 设置过期时间
    r.expire('status:test', 3600)
    print(f"设置1小时后过期的TTL: {r.ttl('status:test')}")
    # 删除key
    r.delete('status:test')
    print(f"删除后的TTL: {r.ttl('status:test')}")
def demo_hash_with_expire():
    """哈希类型设置过期时间示例"""
    print("\n=== 哈希类型过期示例 ===")
    # 设置hash
    r.hset('hash:user:1001', mapping={
        'name': 'Bob',
        'age': '25',
        'city': 'Beijing'
    })
    # 对整个hash设置过期时间
    r.expire('hash:user:1001', 60)
    print(f"Hash过期时间: {r.ttl('hash:user:1001')}秒")
    # 注意：不能对hash中的单个字段设置过期时间
def demo_list_with_expire():
    """列表类型设置过期时间示例"""
    print("\n=== 列表类型过期示例 ===")
    # 设置列表
    r.rpush('list:queue', 'task1', 'task2', 'task3')
    # 对整个列表设置过期时间
    r.expire('list:queue', 30)
    print(f"列表过期时间: {r.ttl('list:queue')}秒")
# 更实用的业务场景示例
def demo_session_management():
    """会话管理示例"""
    print("\n=== 会话管理示例 ===")
    def create_session(user_id, token):
        """创建会话，30分钟过期"""
        session_key = f"session:{token}"
        session_data = {
            'user_id': user_id,
            'login_time': datetime.now().isoformat(),
            'ip': '192.168.1.1'
        }
        # 设置会话数据，30分钟过期
        r.hset(session_key, mapping=session_data)
        r.expire(session_key, 1800)  # 30分钟
        print(f"会话 {token} 创建成功，30分钟后过期")
    def validate_session(token):
        """验证会话是否有效"""
        session_key = f"session:{token}"
        if r.exists(session_key):
            # 会话有效，刷新过期时间（滑动过期）
            r.expire(session_key, 1800)  # 重新设置为30分钟
            print(f"会话 {token} 有效，已刷新过期时间")
            return True
        else:
            print(f"会话 {token} 已过期或不存在")
            return False
    # 测试
    create_session('user_001', 'token_abc123')
    time.sleep(2)
    validate_session('token_abc123')
    time.sleep(2)
    # 删除会话模拟退出登录
    r.delete('session:token_abc123')
    validate_session('token_abc123')
def demo_rate_limiting():
    """速率限制示例"""
    print("\n=== 速率限制示例 ===")
    def check_rate_limit(user_id, limit=5, window=60):
        """
        速率限制检查
        limit: 在window秒内允许的最大请求数
        window: 时间窗口（秒）
        """
        key = f"ratelimit:{user_id}"
        # 获取当前计数
        current = r.get(key)
        if current is None:
            # 第一次请求，设置计数为1，过期时间为window秒
            r.setex(key, window, 1)
            return True, 1, limit
        elif int(current) < limit:
            # 未超过限制，增加计数
            r.incr(key)
            return True, int(current) + 1, limit
        else:
            # 超过限制
            return False, int(current), limit
    # 测试速率限制
    user = 'user_002'
    for i in range(7):
        allowed, count, limit = check_rate_limit(user, limit=5, window=10)
        print(f"请求 {i+1}: {'允许' if allowed else '拒绝'} (当前: {count}/{limit})")
        time.sleep(0.5)
if __name__ == "__main__":
    try:
        # 运行所有示例
        demo_basic_expire()
        demo_expire_with_update()
        demo_remove_expire()
        demo_check_expire_status()
        demo_hash_with_expire()
        demo_list_with_expire()
        demo_session_management()
        demo_rate_limiting()
    except redis.exceptions.ConnectionError as e:
        print(f"Redis连接失败: {e}")
        print("请确保Redis服务已启动")

重要注意事项

# 1. 过期时间的精度问题
# 使用TTL（秒）或PTTL（毫秒），注意选择合适的精度
# 2. 过期时间不能嵌套
# 对已设置过期的key再次设置值，需要重新设置过期时间
r.set('key1', 'value1', ex=10)
r.set('key1', 'value2')  # 这会移除原来的过期时间！
# 正确做法：
r.set('key1', 'value2', ex=10)  # 同时设置值和过期时间
# 3. 批量操作不支持单独设置过期时间
# MSET不支持过期时间，需要逐个设置
pipeline = r.pipeline()
pipeline.set('key1', 'value1', ex=10)
pipeline.set('key2', 'value2', ex=20)
pipeline.execute()
# 4. 检查过期时间
ttl = r.ttl('mykey')  # -1: 永不过期，-2: key不存在，>=0: 剩余秒数
# 5. 删除key会同时删除过期时间
r.delete('mykey')  # 删除key及其过期时间