张芷铭的个人博客

#Python #Redis #缓存 #数据库

Redis 是高性能内存数据库，Python 通过 redis-py 库操作，支持 String、Hash、List、Set、Sorted Set 五种数据结构。

安装与连接

1

pip install redis

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

import redis

# 连接池（推荐）
pool = redis.ConnectionPool(
    host='localhost', port=6379, db=0,
    max_connections=20, decode_responses=True
)
client = redis.Redis(connection_pool=pool)

# 测试连接
if client.ping():
    print("连接成功")

数据结构操作

结构场景写入读取
String缓存、计数器set(key, val)get(key)
Hash对象存储hset(name, k, v)hgetall(name)
List队列lpush/rpushlrange(key, 0, -1)
Set标签、去重sadd(key, *members)smembers(key)
Sorted Set排行榜zadd(key, {m:s})zrevrange(key, 0, -1)

String

1
2
3
4
5

client.set('username', '张三')
client.setex('code', 300, '123456')  # 5分钟后过期
value = client.get('username')

client.incr('page_views')  # 自增

Hash

1
2
3

client.hset('user:1001', 'name', '李四')
client.hset('user:1001', mapping={'age': 30, 'city': '深圳'})
user = client.hgetall('user:1001')

List

1
2

client.lpush('tasks', 'task1', 'task2')
tasks = client.lrange('tasks', 0, -1)

Sorted Set

1
2

client.zadd('rank', {'张三': 95, '李四': 88})
top2 = client.zrevrange('rank', 0, 1, withscores=True)

高级功能

事务（Pipeline）

1
2
3
4
5

with client.pipeline() as pipe:
    pipe.multi()
    pipe.set('key1', 'val1')
    pipe.hset('user:1', 'name', 'test')
    pipe.execute()

分布式锁

1
2
3
4
5
6
7

lock = client.set('my_lock', 'locked', nx=True, ex=10)
if lock:
    try:
        # 执行临界区代码
        pass
    finally:
        client.delete('my_lock')

缓存模式

1
2
3
4
5
6
7

def get_data(key):
    cached = client.get(key)
    if cached:
        return cached
    data = expensive_operation()
    client.setex(key, 3600, data)
    return data

最佳实践

  1. 连接池：避免频繁建立连接
  2. 批量操作：使用 mget/mset 或 pipeline
  3. 异常处理：捕获 ConnectionError、TimeoutError
  4. 过期时间：防止内存溢出

Comments