Python操作Redis从基础到高级应用

pip install redis

pip install redis[asyncio]

import redis

# 创建连接池（推荐）
pool = redis.ConnectionPool(host='localhost', port=6379, db=0, password='yourpassword')
r = redis.Redis(connection_pool=pool)

# 简单连接
r = redis.Redis(host='localhost', port=6379, db=0)

# 设置和获取
r.set('name', 'Alice')
print(r.get('name'))  # 输出: b'Alice'

# 批量操作
r.mset({'key1': 'value1', 'key2': 'value2'})
print(r.mget('key1', 'key2'))  # 输出: [b'value1', b'value2']

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

# 列表操作
r.lpush('tasks', 'task1', 'task2')
r.rpush('tasks', 'task3')
print(r.lrange('tasks', 0, -1))  # 输出: [b'task2', b'task1', b'task3']

# 弹出元素
task = r.lpop('tasks')
print(task)  # 输出: b'task2'

# 哈希表操作
r.hset('user:1000', mapping={
    'name': 'John',
    'age': '30',
    'email': 'john@example.com'
})

print(r.hgetall('user:1000'))  # 输出: {b'name': b'John', b'age': b'30', b'email': b'john@example.com'}

# 获取单个字段
print(r.hget('user:1000', 'name'))  # 输出: b'John'

# 集合操作
r.sadd('tags', 'python', 'redis', 'database')
print(r.smembers('tags'))  # 输出: {b'python', b'redis', b'database'}

# 集合运算
r.sadd('tags2', 'python', 'java')
print(r.sinter('tags', 'tags2'))  # 输出: {b'python'}

# 有序集合
r.zadd('rankings', {'player1': 100, 'player2': 85, 'player3': 95})
print(r.zrevrange('rankings', 0, 1))  # 输出: [b'player1', b'player3']

# 事务示例
pipe = r.pipeline()
pipe.set('tx_key1', 'value1')
pipe.set('tx_key2', 'value2')
pipe.execute()  # 提交事务

# 发布端
r.publish('news', 'Breaking news!')

# 订阅端
pubsub = r.pubsub()
pubsub.subscribe('news')

for message in pubsub.listen():
    if message['type'] == 'message':
        print(f"收到消息: {message['data']}")
        break

# Lua脚本示例
script = """
local current = redis.call('GET', KEYS[1])
local new = current + ARGV[1]
redis.call('SET', KEYS[1], new)
return new
"""

counter = r.eval(script, 1, 'mycounter', 5)
print(counter)  # 输出增量后的值

pool = redis.ConnectionPool(
    max_connections=50,
    host='localhost',
    port=6379,
    decode_responses=True  # 自动解码为字符串
)
r = redis.Redis(connection_pool=pool)

# 管道批量操作
with r.pipeline() as pipe:
    for i in range(1000):
        pipe.set(f'key:{i}', f'value:{i}')
    pipe.execute()  # 一次性提交所有命令

def redis_cache(ttl=60):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # 生成唯一缓存键
            cache_key = f"{func.__name__}:{args}:{frozenset(kwargs.items())}"
            
            # 尝试从缓存获取
            cached = r.get(cache_key)
            if cached is not None:
                return json.loads(cached)
                
            # 缓存未命中，执行函数
            result = func(*args, **kwargs)
            
            # 设置缓存
            r.setex(cache_key, ttl, json.dumps(result))
            return result
        return wrapper
    return decorator

# 使用示例
@redis_cache(ttl=300)
def get_user_profile(user_id):
    # 模拟数据库查询
    time.sleep(2)
    return {"id": user_id, "name": f"User{user_id}", "score": 85}

from redis.exceptions import TimeoutError

try:
    r.ping()
except TimeoutError:
    print("Redis连接超时")
    # 重连逻辑
    r = redis.Redis(host='localhost', socket_timeout=5)

# 检测大Key
big_keys = r.execute_command('MEMORY USAGE', 'some_large_key')
if big_keys > 1024 * 1024:  # 大于1MB
    print(f"警告: 大Key detected - {big_keys} bytes")

from redis.cluster import RedisCluster

rc = RedisCluster(
    startup_nodes=[
        {"host": "127.0.0.1", "port": "7000"},
        {"host": "127.0.0.1", "port": "7001"}
    ],
    decode_responses=True
)
rc.set("cluster_key", "value")