Redis与Python集成详解

1. Redis与Python集成概述

Redis是一种高性能的内存数据库，而Python是一种广泛使用的高级编程语言。将Redis与Python集成，可以充分利用Redis的高性能特性，为Python应用提供更强大的数据处理能力。

1.1 Redis与Python集成的优势

• 高性能缓存：使用Redis作为Python应用的缓存，提高数据访问速度
• 数据持久化：Redis支持多种持久化方式，确保数据安全
• 丰富的数据结构：Redis提供多种数据结构，满足不同场景需求
• 分布式能力：Redis支持主从复制、哨兵模式和集群，提供高可用性
• 易于使用：Python的Redis客户端库提供了简洁易用的API
• 广泛的社区支持：Redis和Python都有活跃的社区，提供丰富的资源和支持

1.2 Redis与Python集成的应用场景

• Web应用缓存：缓存热点数据，减轻数据库压力
• 会话管理：存储用户会话数据，实现会话共享
• 任务队列：实现异步任务处理
• 实时计数器：统计网站访问量、用户在线数等
• 发布/订阅系统：实现消息通知、事件处理
• 地理位置服务：基于Redis的Geo数据类型实现位置相关功能
• 分布式锁：实现分布式系统中的并发控制

2. Redis Python客户端库

2.1 redis-py库

redis-py是Redis官方推荐的Python客户端库，提供了全面的Redis命令支持和友好的Python API。

2.1.1 安装redis-py

pip install redis

2.1.2 基本用法

import redis

# 连接Redis
r = redis.Redis(
    host='localhost',
    port=6379,
    db=0,
    password=None,  # 如果Redis有密码，设置密码
    decode_responses=True  # 自动将响应解码为字符串
)

# 测试连接
print(r.ping())  # 输出: True

# 设置键值对
r.set('name', 'Redis')

# 获取值
print(r.get('name'))  # 输出: Redis

# 删除键
r.delete('name')

# 检查键是否存在
print(r.exists('name'))  # 输出: 0 (False)

2.2 redis-py-cluster库

redis-py-cluster是用于连接Redis Cluster的Python客户端库，基于redis-py开发。

2.2.1 安装redis-py-cluster

pip install redis-py-cluster

2.2.2 基本用法

from rediscluster import RedisCluster

# Redis Cluster节点
startup_nodes = [
    {'host': '127.0.0.1', 'port': '7000'},
    {'host': '127.0.0.1', 'port': '7001'},
    {'host': '127.0.0.1', 'port': '7002'}
]

# 连接Redis Cluster
rc = RedisCluster(
    startup_nodes=startup_nodes,
    decode_responses=True
)

# 设置键值对
rc.set('name', 'Redis Cluster')

# 获取值
print(rc.get('name'))  # 输出: Redis Cluster

2.3 其他Python Redis客户端库

• walrus：一个高级Redis客户端库，提供了对象映射功能
• aredis：一个异步Redis客户端库，基于asyncio
• aioredis：另一个异步Redis客户端库，提供了异步API

3. Redis与Python核心操作

3.1 字符串操作

import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# 设置字符串
r.set('name', 'Redis')
r.setex('expire_key', 10, 'value')  # 10秒后过期
r.mset({'key1': 'value1', 'key2': 'value2'})  # 批量设置

# 获取字符串
print(r.get('name'))  # 输出: Redis
print(r.mget('key1', 'key2'))  # 输出: ['value1', 'value2']

# 字符串操作
r.append('name', ' Python')  # 追加字符串
print(r.get('name'))  # 输出: Redis Python

print(r.strlen('name'))  # 输出: 12

# 数值操作
r.set('counter', 1)
r.incr('counter')  # 自增1
print(r.get('counter'))  # 输出: 2

r.decr('counter', 2)  # 自减2
print(r.get('counter'))  # 输出: 0

3.2 列表操作

import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# 列表操作
r.lpush('mylist', 'a', 'b', 'c')  # 左侧插入
r.rpush('mylist', 'd', 'e')  # 右侧插入

print(r.lrange('mylist', 0, -1))  # 输出: ['c', 'b', 'a', 'd', 'e']

# 弹出元素
print(r.lpop('mylist'))  # 输出: c
print(r.rpop('mylist'))  # 输出: e

print(r.lrange('mylist', 0, -1))  # 输出: ['b', 'a', 'd']

# 列表长度
print(r.llen('mylist'))  # 输出: 3

# 插入元素
r.linsert('mylist', 'before', 'a', 'x')  # 在'a'前插入'x'
print(r.lrange('mylist', 0, -1))  # 输出: ['b', 'x', 'a', 'd']

# 设置元素
r.lset('mylist', 1, 'y')  # 将索引1的元素设置为'y'
print(r.lrange('mylist', 0, -1))  # 输出: ['b', 'y', 'a', 'd']

# 删除元素
r.lrem('mylist', 1, 'a')  # 删除1个'a'
print(r.lrange('mylist', 0, -1))  # 输出: ['b', 'y', 'd']

3.3 集合操作

import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# 集合操作
r.sadd('myset', 'a', 'b', 'c')  # 添加元素

print(r.smembers('myset'))  # 输出: {'a', 'b', 'c'}

# 检查元素是否存在
print(r.sismember('myset', 'a'))  # 输出: True
print(r.sismember('myset', 'd'))  # 输出: False

# 集合长度
print(r.scard('myset'))  # 输出: 3

# 删除元素
r.srem('myset', 'c')  # 删除元素'c'
print(r.smembers('myset'))  # 输出: {'a', 'b'}

# 随机元素
print(r.srandmember('myset'))  # 输出: 随机元素
print(r.spop('myset'))  # 输出: 弹出并返回随机元素

# 集合运算
r.sadd('set1', 'a', 'b', 'c')
r.sadd('set2', 'b', 'c', 'd')

print(r.sinter('set1', 'set2'))  # 交集: {'b', 'c'}
print(r.sunion('set1', 'set2'))  # 并集: {'a', 'b', 'c', 'd'}
print(r.sdiff('set1', 'set2'))  # 差集: {'a'}

3.4 哈希操作

import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# 哈希操作
r.hset('user:1', 'name', 'Alice')
r.hset('user:1', 'age', 30)
r.hset('user:1', 'email', 'alice@example.com')

# 批量设置
data = {
    'name': 'Bob',
    'age': 25,
    'email': 'bob@example.com'
}
r.hset('user:2', mapping=data)

# 获取字段
print(r.hget('user:1', 'name'))  # 输出: Alice
print(r.hmget('user:1', 'name', 'age'))  # 输出: ['Alice', '30']

# 获取所有字段和值
print(r.hgetall('user:1'))  # 输出: {'name': 'Alice', 'age': '30', 'email': 'alice@example.com'}

# 获取所有字段
print(r.hkeys('user:1'))  # 输出: ['name', 'age', 'email']

# 获取所有值
print(r.hvals('user:1'))  # 输出: ['Alice', '30', 'alice@example.com']

# 哈希长度
print(r.hlen('user:1'))  # 输出: 3

# 检查字段是否存在
print(r.hexists('user:1', 'name'))  # 输出: True
print(r.hexists('user:1', 'address'))  # 输出: False

# 删除字段
r.hdel('user:1', 'email')  # 删除'email'字段
print(r.hgetall('user:1'))  # 输出: {'name': 'Alice', 'age': '30'}

# 数值操作
r.hincrby('user:1', 'age', 1)  # 年龄自增1
print(r.hget('user:1', 'age'))  # 输出: 31

3.5 有序集合操作

import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# 有序集合操作
r.zadd('scores', {'Alice': 85, 'Bob': 92, 'Charlie': 78, 'David': 95})

# 获取元素分数
print(r.zscore('scores', 'Alice'))  # 输出: 85.0

# 按分数范围获取元素
print(r.zrange('scores', 0, -1, withscores=True))  # 升序
print(r.zrevrange('scores', 0, -1, withscores=True))  # 降序

# 按分数范围获取
print(r.zrangebyscore('scores', 80, 90, withscores=True))  # 分数在80-90之间

# 元素排名
print(r.zrank('scores', 'Alice'))  # 升序排名 (0开始)
print(r.zrevrank('scores', 'Alice'))  # 降序排名 (0开始)

# 有序集合长度
print(r.zcard('scores'))  # 输出: 4

# 删除元素
r.zrem('scores', 'Charlie')  # 删除'Charlie'
print(r.zrange('scores', 0, -1, withscores=True))

# 分数操作
r.zincrby('scores', 5, 'Alice')  # Alice的分数增加5
print(r.zscore('scores', 'Alice'))  # 输出: 90.0

# 统计分数范围内的元素数量
print(r.zcount('scores', 90, 100))  # 输出: 3

4. Redis Python客户端高级特性

4.1 连接池

连接池是管理Redis连接的重要机制，可以减少连接建立和关闭的开销，提高性能。

import redis

# 创建连接池
pool = redis.ConnectionPool(
    host='localhost',
    port=6379,
    db=0,
    max_connections=50,  # 最大连接数
    decode_responses=True
)

# 使用连接池
r = redis.Redis(connection_pool=pool)

# 测试连接
print(r.ping())  # 输出: True

# 执行操作
r.set('key', 'value')
print(r.get('key'))  # 输出: value

4.2 管道

管道可以批量执行多个Redis命令，减少网络往返时间，提高性能。

import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# 使用管道
pipe = r.pipeline()

# 批量添加命令
pipe.set('key1', 'value1')
pipe.set('key2', 'value2')
pipe.get('key1')
pipe.get('key2')

# 执行所有命令
results = pipe.execute()
print(results)  # 输出: [True, True, 'value1', 'value2']

# 事务性管道 (默认情况下管道是原子的)
pipe = r.pipeline(transaction=True)
try:
    pipe.set('user:1:balance', 100)
    pipe.decrby('user:1:balance', 50)
    pipe.get('user:1:balance')
    results = pipe.execute()
    print(results)  # 输出: [True, 50, '50']
except Exception as e:
    print(f"事务执行失败: {e}")
    pipe.reset()

4.3 发布/订阅

Redis的发布/订阅功能可以用于实现消息通知、事件处理等场景。

# 发布者示例
import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# 发布消息
r.publish('channel1', 'Hello Redis!')
r.publish('channel1', 'How are you?')

# 订阅者示例
import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# 创建订阅对象
pubsub = r.pubsub()

# 订阅频道
pubsub.subscribe('channel1')

# 接收消息
print("开始接收消息...")
for message in pubsub.listen():
    if message['type'] == 'message':
        print(f"接收到消息: {message['data']}")
    elif message['type'] == 'subscribe':
        print(f"订阅了频道: {message['channel']}")
    # 可以添加条件来退出循环
    # if some_condition:
    #     break

# 取消订阅
pubsub.unsubscribe('channel1')

4.4 Lua脚本

Redis支持执行Lua脚本，可以在服务器端原子性地执行复杂操作。

import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# 定义Lua脚本
lua_script = """
    local balance = redis.call('get', KEYS[1])
    if not balance then
        return -1
    end
    balance = tonumber(balance)
    if balance < tonumber(ARGV[1]) then
        return 0
    end
    redis.call('decrby', KEYS[1], ARGV[1])
    return 1
"""

# 加载脚本
script = r.register_script(lua_script)

# 设置初始余额
r.set('user:1:balance', 100)

# 执行脚本
result = script(keys=['user:1:balance'], args=['60'])
print(f"执行结果: {result}")  # 输出: 1
print(f"剩余余额: {r.get('user:1:balance')}")  # 输出: 40

# 尝试执行一个会失败的操作
result = script(keys=['user:1:balance'], args=['50'])
print(f"执行结果: {result}")  # 输出: 0 (余额不足)
print(f"剩余余额: {r.get('user:1:balance')}")  # 输出: 40

5. 实用案例分析

5.1 缓存装饰器

5.1.1 需求分析

• 创建一个通用的缓存装饰器，用于缓存函数的返回值
• 支持设置缓存过期时间
• 支持缓存键的自定义生成
• 支持缓存的清除

5.1.2 实现方案

import redis
import functools
import hashlib
import json

# 连接Redis
r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# 缓存装饰器
def redis_cache(expire=3600, key_prefix='cache'):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # 生成缓存键
            key_parts = [key_prefix, func.__name__]
            
            # 添加位置参数
            for arg in args:
                if isinstance(arg, (int, str, float, bool)):
                    key_parts.append(str(arg))
                elif isinstance(arg, (list, dict, tuple)):
                    # 对复杂类型进行哈希
                    hash_obj = hashlib.md5(json.dumps(arg, sort_keys=True).encode())
                    key_parts.append(hash_obj.hexdigest())
            
            # 添加关键字参数
            for k, v in sorted(kwargs.items()):
                key_parts.append(f"{k}:{v}")
            
            # 生成最终的缓存键
            cache_key = ':'.join(key_parts)
            
            # 尝试从缓存获取
            cached_value = r.get(cache_key)
            if cached_value:
                print(f"从缓存获取: {cache_key}")
                return json.loads(cached_value)
            
            # 执行函数
            result = func(*args, **kwargs)
            
            # 将结果存入缓存
            print(f"存入缓存: {cache_key}")
            r.setex(cache_key, expire, json.dumps(result))
            
            return result
        
        # 添加清除缓存的方法
        def clear_cache(*args, **kwargs):
            # 生成与wrapper相同的缓存键
            key_parts = [key_prefix, func.__name__]
            
            # 添加位置参数
            for arg in args:
                if isinstance(arg, (int, str, float, bool)):
                    key_parts.append(str(arg))
                elif isinstance(arg, (list, dict, tuple)):
                    # 对复杂类型进行哈希
                    hash_obj = hashlib.md5(json.dumps(arg, sort_keys=True).encode())
                    key_parts.append(hash_obj.hexdigest())
            
            # 添加关键字参数
            for k, v in sorted(kwargs.items()):
                key_parts.append(f"{k}:{v}")
            
            # 生成最终的缓存键
            cache_key = ':'.join(key_parts)
            
            # 删除缓存
            r.delete(cache_key)
            print(f"清除缓存: {cache_key}")
        
        wrapper.clear_cache = clear_cache
        return wrapper
    return decorator

# 测试缓存装饰器
@redis_cache(expire=60, key_prefix='example')
def expensive_function(a, b, c=None):
    print(f"执行昂贵的函数: a={a}, b={b}, c={c}")
    # 模拟耗时操作
    import time
    time.sleep(1)
    return a + b + (c if c else 0)

# 第一次调用，会执行函数
print("第一次调用:")
result1 = expensive_function(1, 2, c=3)
print(f"结果: {result1}")

# 第二次调用，会从缓存获取
print("\n第二次调用:")
result2 = expensive_function(1, 2, c=3)
print(f"结果: {result2}")

# 清除缓存
print("\n清除缓存:")
expensive_function.clear_cache(1, 2, c=3)

# 第三次调用，会重新执行函数
print("\n第三次调用:")
result3 = expensive_function(1, 2, c=3)
print(f"结果: {result3}")

# 不同参数的调用，会执行函数
print("\n不同参数的调用:")
result4 = expensive_function(4, 5)
print(f"结果: {result4}")

5.1.3 运行结果

第一次调用:
执行昂贵的函数: a=1, b=2, c=3
存入缓存: example:expensive_function:1:2:c:3
结果: 6

第二次调用:
从缓存获取: example:expensive_function:1:2:c:3
结果: 6

清除缓存:
清除缓存: example:expensive_function:1:2:c:3

第三次调用:
执行昂贵的函数: a=1, b=2, c=3
存入缓存: example:expensive_function:1:2:c:3
结果: 6

不同参数的调用:
执行昂贵的函数: a=4, b=5, c=None
存入缓存: example:expensive_function:4:5
结果: 9

5.2 会话管理

5.2.1 需求分析

• 使用Redis存储用户会话数据
• 支持会话的创建、获取、更新和删除
• 支持会话过期
• 提供简洁的API接口

5.2.2 实现方案

import redis
import uuid
import json
import time

class RedisSession:
    def __init__(self, redis_client, prefix='session', expire=3600):
        """
        初始化Redis会话管理
        :param redis_client: Redis客户端实例
        :param prefix: 会话键前缀
        :param expire: 会话过期时间（秒）
        """
        self.redis = redis_client
        self.prefix = prefix
        self.expire = expire
    
    def create_session(self, user_data):
        """
        创建新会话
        :param user_data: 用户数据字典
        :return: 会话ID
        """
        session_id = str(uuid.uuid4())
        session_key = f"{self.prefix}:{session_id}"
        
        # 添加会话创建时间
        session_data = {
            'user_data': user_data,
            'created_at': int(time.time()),
            'last_accessed': int(time.time())
        }
        
        # 存储会话数据
        self.redis.setex(session_key, self.expire, json.dumps(session_data))
        return session_id
    
    def get_session(self, session_id):
        """
        获取会话数据
        :param session_id: 会话ID
        :return: 会话数据字典，如果会话不存在或已过期返回None
        """
        session_key = f"{self.prefix}:{session_id}"
        session_data = self.redis.get(session_key)
        
        if not session_data:
            return None
        
        # 更新最后访问时间
        session_data = json.loads(session_data)
        session_data['last_accessed'] = int(time.time())
        self.redis.setex(session_key, self.expire, json.dumps(session_data))
        
        return session_data
    
    def update_session(self, session_id, user_data):
        """
        更新会话数据
        :param session_id: 会话ID
        :param user_data: 更新的用户数据字典
        :return: 是否更新成功
        """
        session_key = f"{self.prefix}:{session_id}"
        session_data = self.redis.get(session_key)
        
        if not session_data:
            return False
        
        # 更新会话数据
        session_data = json.loads(session_data)
        session_data['user_data'] = user_data
        session_data['last_accessed'] = int(time.time())
        
        self.redis.setex(session_key, self.expire, json.dumps(session_data))
        return True
    
    def delete_session(self, session_id):
        """
        删除会话
        :param session_id: 会话ID
        :return: 是否删除成功
        """
        session_key = f"{self.prefix}:{session_id}"
        result = self.redis.delete(session_key)
        return result > 0
    
    def get_user_data(self, session_id):
        """
        获取会话中的用户数据
        :param session_id: 会话ID
        :return: 用户数据字典，如果会话不存在或已过期返回None
        """
        session_data = self.get_session(session_id)
        if session_data:
            return session_data.get('user_data')
        return None

# 测试会话管理
# 连接Redis
r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# 创建会话管理器
session_manager = RedisSession(r, expire=30)  # 30秒过期

# 测试1: 创建会话
print("测试1: 创建会话")
user_data = {'user_id': 1, 'username': 'alice', 'email': 'alice@example.com'}
session_id = session_manager.create_session(user_data)
print(f"创建会话成功，会话ID: {session_id}")

# 测试2: 获取会话
print("\n测试2: 获取会话")
session_data = session_manager.get_session(session_id)
print(f"获取会话成功: {session_data}")

# 测试3: 获取用户数据
print("\n测试3: 获取用户数据")
user_data_from_session = session_manager.get_user_data(session_id)
print(f"获取用户数据成功: {user_data_from_session}")

# 测试4: 更新会话
print("\n测试4: 更新会话")
updated_user_data = {'user_id': 1, 'username': 'alice', 'email': 'alice.new@example.com', 'age': 30}
success = session_manager.update_session(session_id, updated_user_data)
print(f"更新会话成功: {success}")

# 验证更新
user_data_from_session = session_manager.get_user_data(session_id)
print(f"更新后的用户数据: {user_data_from_session}")

# 测试5: 删除会话
print("\n测试5: 删除会话")
success = session_manager.delete_session(session_id)
print(f"删除会话成功: {success}")

# 验证删除
session_data = session_manager.get_session(session_id)
print(f"删除后会话是否存在: {session_data is not None}")

# 测试6: 会话过期
print("\n测试6: 会话过期")
# 创建新会话
session_id2 = session_manager.create_session({'user_id': 2, 'username': 'bob'})
print(f"创建新会话成功，会话ID: {session_id2}")

# 等待会话过期
print("等待35秒让会话过期...")
time.sleep(35)

# 尝试获取过期的会话
session_data = session_manager.get_session(session_id2)
print(f"获取过期会话: {session_data}")

5.2.3 运行结果

测试1: 创建会话
创建会话成功，会话ID: 550e8400-e29b-41d4-a716-446655440000

测试2: 获取会话
获取会话成功: {'user_data': {'user_id': 1, 'username': 'alice', 'email': 'alice@example.com'}, 'created_at': 1620000000, 'last_accessed': 1620000000}

测试3: 获取用户数据
获取用户数据成功: {'user_id': 1, 'username': 'alice', 'email': 'alice@example.com'}

测试4: 更新会话
更新会话成功: True
更新后的用户数据: {'user_id': 1, 'username': 'alice', 'email': 'alice.new@example.com', 'age': 30}

测试5: 删除会话
删除会话成功: True
删除后会话是否存在: False

测试6: 会话过期
创建新会话成功，会话ID: b1b4b040-4e4f-4f29-9c15-3f8d48a5e210
等待35秒让会话过期...
获取过期会话: None

5.3 任务队列

5.3.1 需求分析

• 使用Redis实现一个简单的任务队列
• 支持任务的提交、获取和处理
• 支持任务的优先级
• 支持任务的重试机制
• 提供简洁的API接口

5.3.2 实现方案

import redis
import json
import time
import uuid

class RedisTaskQueue:
    def __init__(self, redis_client, queue_name='task_queue', retry_queue_name='retry_queue'):
        """
        初始化任务队列
        :param redis_client: Redis客户端实例
        :param queue_name: 任务队列名称
        :param retry_queue_name: 重试队列名称
        """
        self.redis = redis_client
        self.queue_name = queue_name
        self.retry_queue_name = retry_queue_name
    
    def enqueue(self, task_type, task_data, priority=0, max_retries=3):
        """
        提交任务到队列
        :param task_type: 任务类型
        :param task_data: 任务数据
        :param priority: 任务优先级，数字越大优先级越高
        :param max_retries: 最大重试次数
        :return: 任务ID
        """
        task_id = str(uuid.uuid4())
        task = {
            'task_id': task_id,
            'task_type': task_type,
            'task_data': task_data,
            'priority': priority,
            'created_at': int(time.time()),
            'retries': 0,
            'max_retries': max_retries
        }
        
        # 使用有序集合存储任务，按优先级排序
        # 分数使用负数，因为Redis有序集合是按分数升序排列的
        score = -priority
        self.redis.zadd(self.queue_name, {json.dumps(task): score})
        return task_id
    
    def dequeue(self, block=False, timeout=0):
        """
        从队列获取任务
        :param block: 是否阻塞等待
        :param timeout: 阻塞超时时间（秒）
        :return: 任务字典，如果队列为空返回None
        """
        if block:
            # 阻塞方式获取任务
            # 这里使用一种简单的轮询方式，实际生产环境可以使用Redis的BLPOP等命令
            start_time = time.time()
            while time.time() - start_time < timeout:
                task = self._dequeue_one()
                if task:
                    return task
                time.sleep(0.1)
            return None
        else:
            # 非阻塞方式获取任务
            return self._dequeue_one()
    
    def _dequeue_one(self):
        """
        内部方法：获取一个任务
        :return: 任务字典，如果队列为空返回None
        """
        # 获取优先级最高的任务（分数最小的，因为我们使用了负数）
        result = self.redis.zpopmin(self.queue_name)
        if result:
            task_json = result[0][0]
            return json.loads(task_json)
        return None
    
    def requeue(self, task, delay=0):
        """
        将任务重新加入队列（用于重试）
        :param task: 任务字典
        :param delay: 延迟时间（秒）
        :return: 是否成功
        """
        # 检查重试次数
        if task['retries'] >= task['max_retries']:
            print(f"任务 {task['task_id']} 已达到最大重试次数，不再重试")
            return False
        
        # 增加重试次数
        task['retries'] += 1
        task['last_retried_at'] = int(time.time())
        
        if delay > 0:
            # 延迟重试，使用另一个队列存储
            retry_key = f"{self.retry_queue_name}:{int(time.time()) + delay}"
            self.redis.lpush(retry_key, json.dumps(task))
            # 设置过期时间，避免键永久存在
            self.redis.expire(retry_key, delay + 60)  # 额外60秒缓冲
        else:
            # 立即重试，重新加入主队列
            score = -task['priority']
            self.redis.zadd(self.queue_name, {json.dumps(task): score})
        
        return True
    
    def queue_size(self):
        """
        获取队列大小
        :return: 队列中的任务数量
        """
        return self.redis.zcard(self.queue_name)
    
    def process_retry_queue(self):
        """
        处理延迟重试队列
        :return: 处理的任务数量
        """
        current_time = int(time.time())
        processed = 0
        
        # 查找所有过期的重试键
        retry_keys = self.redis.keys(f"{self.retry_queue_name}:*")
        for key in retry_keys:
            # 提取时间戳
            timestamp_str = key.split(':')[-1]
            try:
                timestamp = int(timestamp_str)
                if timestamp <= current_time:
                    # 处理该键中的所有任务
                    while True:
                        task_json = self.redis.lpop(key)
                        if not task_json:
                            break
                        task = json.loads(task_json)
                        # 将任务重新加入主队列
                        score = -task['priority']
                        self.redis.zadd(self.queue_name, {json.dumps(task): score})
                        processed += 1
            except ValueError:
                pass
        
        return processed

# 测试任务队列
# 连接Redis
r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# 创建任务队列
task_queue = RedisTaskQueue(r)

# 测试1: 清空队列（如果有旧数据）
r.delete(task_queue.queue_name)
r.delete(*r.keys(f"{task_queue.retry_queue_name}:*"))

# 测试2: 提交任务
print("测试2: 提交任务")
task_ids = []

# 提交低优先级任务
task_id1 = task_queue.enqueue('send_email', {'to': 'user1@example.com', 'subject': 'Hello'}, priority=1)
task_ids.append(task_id1)
print(f"提交低优先级任务成功，任务ID: {task_id1}")

# 提交高优先级任务
task_id2 = task_queue.enqueue('send_email', {'to': 'user2@example.com', 'subject': 'Important'}, priority=5)
task_ids.append(task_id2)
print(f"提交高优先级任务成功，任务ID: {task_id2}")

# 提交中优先级任务
task_id3 = task_queue.enqueue('send_email', {'to': 'user3@example.com', 'subject': 'Notice'}, priority=3)
task_ids.append(task_id3)
print(f"提交中优先级任务成功，任务ID: {task_id3}")

# 查看队列大小
print(f"队列大小: {task_queue.queue_size()}")

# 测试3: 获取任务（应该按优先级顺序）
print("\n测试3: 获取任务")
for i in range(3):
    task = task_queue.dequeue()
    if task:
        print(f"获取任务 {i+1}: 优先级={task['priority']}, 类型={task['task_type']}, 数据={task['task_data']}")

# 查看队列大小
print(f"队列大小: {task_queue.queue_size()}")

# 测试4: 任务重试
print("\n测试4: 任务重试")
# 提交一个任务
test_task = {
    'task_id': 'test_retry_task',
    'task_type': 'process_data',
    'task_data': {'data': 'test'},
    'priority': 1,
    'created_at': int(time.time()),
    'retries': 0,
    'max_retries': 3
}

# 将任务加入队列
score = -test_task['priority']
r.zadd(task_queue.queue_name, {json.dumps(test_task): score})

# 获取任务
task = task_queue.dequeue()
print(f"获取任务: {task['task_id']}, 当前重试次数: {task['retries']}")

# 模拟处理失败，重新加入队列
print("模拟处理失败，重新加入队列")
success = task_queue.requeue(task)
print(f"重新加入队列成功: {success}")

# 再次获取任务
task = task_queue.dequeue()
print(f"再次获取任务: {task['task_id']}, 当前重试次数: {task['retries']}")

# 测试5: 处理延迟重试
print("\n测试5: 处理延迟重试")
# 重新加入队列，延迟2秒
success = task_queue.requeue(task, delay=2)
print(f"延迟重新加入队列成功: {success}")

# 立即尝试获取任务（应该获取不到）
task = task_queue.dequeue()
print(f"立即尝试获取任务: {task is not None}")

# 处理重试队列（此时应该处理不了，因为延迟时间未到）
processed = task_queue.process_retry_queue()
print(f"处理重试队列，处理任务数: {processed}")

# 等待3秒
print("等待3秒...")
time.sleep(3)

# 再次处理重试队列
processed = task_queue.process_retry_queue()
print(f"处理重试队列，处理任务数: {processed}")

# 现在应该可以获取到任务了
task = task_queue.dequeue()
print(f"获取延迟重试的任务: {task is not None}")
if task:
    print(f"任务信息: {task['task_id']}, 重试次数: {task['retries']}")

# 查看队列大小
print(f"最终队列大小: {task_queue.queue_size()}")

5.3.3 运行结果

测试2: 提交任务
提交低优先级任务成功，任务ID: 550e8400-e29b-41d4-a716-446655440000
提交高优先级任务成功，任务ID: b1b4b040-4e4f-4f29-9c15-3f8d48a5e210
提交中优先级任务成功，任务ID: 1a2b3c4d-5e6f-7g8h-9i0j-k1l2m3n4o5p6
队列大小: 3

测试3: 获取任务
获取任务 1: 优先级=5, 类型=send_email, 数据={'to': 'user2@example.com', 'subject': 'Important'}
获取任务 2: 优先级=3, 类型=send_email, 数据={'to': 'user3@example.com', 'subject': 'Notice'}
获取任务 3: 优先级=1, 类型=send_email, 数据={'to': 'user1@example.com', 'subject': 'Hello'}
队列大小: 0

测试4: 任务重试
获取任务: test_retry_task, 当前重试次数: 0
模拟处理失败，重新加入队列
重新加入队列成功: True
再次获取任务: test_retry_task, 当前重试次数: 1

测试5: 处理延迟重试
延迟重新加入队列成功: True
立即尝试获取任务: False
处理重试队列，处理任务数: 0
等待3秒...
处理重试队列，处理任务数: 1
获取延迟重试的任务: True
任务信息: test_retry_task, 重试次数: 1
最终队列大小: 0

6. Redis与Python集成最佳实践

6.1 性能优化

1. 使用连接池：连接池可以减少连接建立和关闭的开销，提高性能
2. 批量操作：使用管道（pipeline）批量执行命令，减少网络往返时间
3. 合理使用数据结构：根据实际场景选择合适的Redis数据结构
4. 设置合理的过期时间：为缓存数据设置合适的过期时间，避免内存溢出
5. 使用Lua脚本：对于复杂操作，使用Lua脚本在服务器端原子性执行
6. 监控内存使用：定期监控Redis的内存使用情况，避免内存不足
7. 数据压缩：对于大型数据，可以考虑在存储前进行压缩

6.2 可靠性保障

1. 错误处理：添加适当的错误处理机制，处理Redis连接失败等异常
2. 重试机制：对于临时错误，实现重试机制
3. 数据备份：启用Redis的持久化功能，定期备份数据
4. 高可用性：使用Redis的主从复制、哨兵模式或集群，提高系统可用性
5. 监控和告警：设置Redis的监控和告警机制，及时发现问题

6.3 代码质量

1. 封装Redis操作：将Redis操作封装成服务或工具类，提高代码复用性
2. 使用类型提示：为Python代码添加类型提示，提高代码可读性和可维护性
3. 编写单元测试：为Redis相关代码编写单元测试，确保功能正确
4. 文档和注释：添加适当的文档和注释，说明代码的功能和使用方法
5. 代码风格：遵循Python的代码风格规范（如PEP 8）

6.4 安全考虑

1. 密码认证：为Redis设置密码，避免未授权访问
2. 网络隔离：将Redis部署在内部网络，避免暴露在公网
3. 数据加密：对于敏感数据，在存储前进行加密
4. 访问控制：使用Redis的ACL功能，限制用户的操作权限
5. 避免存储敏感信息：尽量避免在Redis中存储明文的敏感信息

7. 常见问题与解决方案

7.1 连接问题

问题：Python应用无法连接到Redis服务器

解决方案：

• 检查Redis服务器是否运行
• 检查网络连接是否正常
• 检查Redis配置中的bind和protected-mode设置
• 检查密码是否正确
• 检查防火墙设置

7.2 内存问题

问题：Redis内存使用过高

解决方案：

• 设置合理的过期时间
• 使用合适的数据结构
• 考虑使用Redis的内存淘汰策略
• 监控内存使用情况，及时扩容

7.3 性能问题

问题：Redis操作性能下降

解决方案：

• 使用连接池
• 批量执行命令
• 优化数据结构
• 考虑使用Redis集群
• 检查服务器资源使用情况

7.4 数据一致性问题

问题：Redis数据与数据库数据不一致

解决方案：

• 实现合适的缓存更新策略（如过期时间、主动更新）
• 使用事务或Lua脚本保证操作的原子性
• 考虑使用消息队列确保数据最终一致性

7.5 序列化问题

问题：Python对象无法直接存储到Redis

解决方案：

• 使用json、pickle等库进行序列化和反序列化
• 对于复杂对象，考虑使用更高效的序列化库（如msgpack）
• 注意序列化的性能和安全性

8. 总结

Redis与Python的集成是一种强大的组合，可以为Python应用提供高性能的数据处理能力。通过本教程的学习，我们了解了Redis与Python集成的方法、常用库、核心操作以及实际应用场景。

8.1 核心知识点回顾

• Redis Python客户端库：redis-py是官方推荐的客户端库，提供了全面的Redis命令支持
• 连接管理：使用连接池管理Redis连接，提高性能
• 数据操作：支持Redis的各种数据结构操作，如字符串、列表、集合、哈希、有序集合
• 高级特性：管道、发布/订阅、Lua脚本等高级特性
• 应用场景：缓存、会话管理、任务队列、计数器等
• 最佳实践：性能优化、可靠性保障、代码质量、安全考虑

8.2 实践建议

1. 从小规模开始：先在小规模应用中使用Redis，熟悉其特性和操作
2. 监控和调优：定期监控Redis的性能和内存使用情况，根据需要进行调优
3. 合理设计数据结构：根据实际场景选择合适的Redis数据结构
4. 考虑高可用性：对于生产环境，使用Redis的高可用方案
5. 学习社区资源：关注Redis和Python的社区，学习最佳实践和解决方案

8.3 未来发展

Redis和Python都在不断发展，未来可能会有更多的集成方式和工具：

• 异步支持：随着Python异步编程的普及，异步Redis客户端库（如aioredis）将更加重要
• 更多的集成工具：可能会出现更多专门的Redis-Python集成工具和框架
• 云服务集成：与云服务提供商的Redis服务更好地集成
• 更高级的功能：Redis的新功能和Python的新特性的结合

通过掌握Redis与Python的集成，开发者可以构建高性能、可靠的应用，满足各种复杂场景的需求。