Python 使用 Redis
Python是Redis最流行的客户端语言之一，本课学习Python操作Redis。
安装redis-py
redis-py是Python最常用的Redis客户端。
安装
BASH
pip install redis
验证安装
PYTHON
import redis
print(redis.__version__) # 输出版本号
基本连接
简单连接
PYTHON
import redis
# 连接Redis
r = redis.Redis(host='localhost', port=6379, db=0)
# 测试连接
print(r.ping()) # True
带密码连接
PYTHON
import redis
r = redis.Redis(
host='localhost',
port=6379,
password='your_password',
db=0
)
连接参数
PYTHON
import redis
r = redis.Redis(
host='localhost', # 主机地址
port=6379, # 端口
password='password', # 密码
db=0, # 数据库编号
decode_responses=True, # 自动解码为字符串
socket_timeout=5, # 超时时间（秒）
socket_connect_timeout=5,
retry_on_timeout=True,
max_connections=10
)
💡 decode_responses： 设置为True后，返回bytes会自动解码为str。
基本操作
字符串操作
PYTHON
import redis
r = redis.Redis(decode_responses=True)
# SET和GET
r.set('name', 'Alice')
print(r.get('name')) # 'Alice'
# SET带过期时间
r.set('session', 'data', ex=3600) # 1小时后过期
# MSET和MGET
r.mset({'key1': 'value1', 'key2': 'value2'})
print(r.mget('key1', 'key2')) # ['value1', 'value2']
# INCR
r.set('counter', 0)
print(r.incr('counter')) # 1
print(r.incrby('counter', 10)) # 11
# APPEND
r.append('name', ' Smith')
print(r.get('name')) # 'Alice Smith'
# STRLEN
print(r.strlen('name')) # 11
# DEL
r.delete('name')
print(r.get('name')) # None
哈希操作
PYTHON
import redis
r = redis.Redis(decode_responses=True)
# HSET和HGET
r.hset('user:1', 'name', 'Alice')
r.hset('user:1', 'age', 25)
print(r.hget('user:1', 'name')) # 'Alice'
# HMSET和HMGET
r.hset('user:2', mapping={'name': 'Bob', 'age': 30, 'city': 'Beijing'})
print(r.hmget('user:2', 'name', 'age')) # ['Bob', '30']
# HGETALL
print(r.hgetall('user:2')) # {'name': 'Bob', 'age': '30', 'city': 'Beijing'}
# HKEYS和HVALS
print(r.hkeys('user:2')) # ['name', 'age', 'city']
print(r.hvals('user:2')) # ['Bob', '30', 'Beijing']
# HDEL
r.hdel('user:2', 'city')
# HINCRBY
r.hincrby('user:2', 'age', 1)
列表操作
PYTHON
import redis
r = redis.Redis(decode_responses=True)
# LPUSH和RPUSH
r.lpush('mylist', 'value1', 'value2')
r.rpush('mylist', 'value3')
# LRANGE
print(r.lrange('mylist', 0, -1)) # ['value2', 'value1', 'value3']
# LPOP和RPOP
print(r.lpop('mylist')) # 'value2'
print(r.rpop('mylist')) # 'value3'
# LLEN
print(r.llen('mylist')) # 1
# LINDEX
print(r.lindex('mylist', 0)) # 'value1'
# LSET
r.lset('mylist', 0, 'new_value')
集合操作
PYTHON
import redis
r = redis.Redis(decode_responses=True)
# SADD
r.sadd('myset', 'a', 'b', 'c')
# SMEMBERS
print(r.smembers('myset')) # {'a', 'b', 'c'}
# SISMEMBER
print(r.sismember('myset', 'a')) # True
# SREM
r.srem('myset', 'a')
# SCARD
print(r.scard('myset')) # 2
# SINTER、SUNION、SDIFF
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'}
有序集合操作
PYTHON
import redis
r = redis.Redis(decode_responses=True)
# ZADD
r.zadd('leaderboard', {'player1': 100, 'player2': 200, 'player3': 150})
# ZRANGE
print(r.zrange('leaderboard', 0, -1, withscores=True))
# [('player1', 100.0), ('player3', 150.0), ('player2', 200.0)]
# ZREVRANGE（从高到低）
print(r.zrevrange('leaderboard', 0, 2, withscores=True))
# ZSCORE
print(r.zscore('leaderboard', 'player1')) # 100.0
# ZRANK和ZREVRANK
print(r.zrank('leaderboard', 'player1')) # 0
print(r.zrevrank('leaderboard', 'player2')) # 0
# ZINCRBY
r.zincrby('leaderboard', 50, 'player1')
# ZREM
r.zrem('leaderboard', 'player1')
连接池
连接池可以复用连接，提高性能。
创建连接池
PYTHON
import redis
# 创建连接池
pool = redis.ConnectionPool(
host='localhost',
port=6379,
db=0,
max_connections=100,
decode_responses=True
)
# 从连接池获取连接
r = redis.Redis(connection_pool=pool)
# 使用连接
r.set('key', 'value')
print(r.get('key'))
连接池优势
- 复用连接，减少连接开销
- 控制最大连接数
- 提高性能
💡 推荐： 生产环境建议使用连接池。
管道操作
PYTHON
import redis
r = redis.Redis(decode_responses=True)
# 创建管道
pipe = r.pipeline()
# 添加命令
pipe.set('key1', 'value1')
pipe.set('key2', 'value2')
pipe.set('key3', 'value3')
pipe.get('key1')
# 执行管道
results = pipe.execute()
print(results) # [True, True, True, 'value1']
# 管道+事务
pipe = r.pipeline(transaction=True)
pipe.set('key1', 'value1')
pipe.set('key2', 'value2')
results = pipe.execute()
发布订阅
发布消息
PYTHON
import redis
r = redis.Redis()
# 发布消息
r.publish('news', 'Hello World')
订阅消息
PYTHON
import redis
r = redis.Redis()
# 订阅频道
pubsub = r.pubsub()
pubsub.subscribe('news')
# 接收消息
for message in pubsub.listen():
if message['type'] == 'message':
print(f"频道：{message['channel']}")
print(f"消息：{message['data']}")
模式订阅
PYTHON
import redis
r = redis.Redis()
pubsub = r.pubsub()
pubsub.psubscribe('news:*')
for message in pubsub.listen():
if message['type'] == 'pmessage':
print(f"模式：{message['pattern']}")
print(f"频道：{message['channel']}")
print(f"消息：{message['data']}")
实战案例
案例1：缓存装饰器
PYTHON
import redis
import json
import functools
r = redis.Redis(decode_responses=True)
def cache(expire=3600):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
# 生成缓存键
cache_key = f"{func.__name__}:{args}:{kwargs}"
# 尝试从缓存获取
cached = r.get(cache_key)
if cached:
return json.loads(cached)
# 执行函数
result = func(*args, **kwargs)
# 存入缓存
r.set(cache_key, json.dumps(result), ex=expire)
return result
return wrapper
return decorator
@cache(expire=300)
def get_user(user_id):
# 模拟数据库查询
print(f"查询数据库：user_id={user_id}")
return {'id': user_id, 'name': f'User{user_id}'}
# 测试
print(get_user(1)) # 查询数据库
print(get_user(1)) # 从缓存获取
案例2：分布式锁
PYTHON
import redis
import time
import uuid
r = redis.Redis()
class DistributedLock:
def __init__(self, key, expire=10):
self.key = f"lock:{key}"
self.expire = expire
self.identifier = str(uuid.uuid4())
def acquire(self):
# 尝试获取锁
return r.set(self.key, self.identifier, nx=True, ex=self.expire)
def release(self):
# 释放锁（使用Lua脚本保证原子性）
script = """
if redis.call("get", KEYS[1]) == ARGV[1] then
return redis.call("del", KEYS[1])
else
return 0
end
"""
return r.eval(script, 1, self.key, self.identifier)
def __enter__(self):
while not self.acquire():
time.sleep(0.1)
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.release()
# 使用锁
with DistributedLock('resource') as lock:
print('执行业务逻辑')
time.sleep(1)
案例3：限流器
PYTHON
import redis
import time
r = redis.Redis()
def rate_limit(key, limit=10, period=60):
"""
限流器
:param key: 限流键
:param limit: 时间窗口内最大请求数
:param period: 时间窗口（秒）
:return: 是否允许
"""
current = int(time.time())
window_start = current - period
# 使用有序集合实现滑动窗口
pipe = r.pipeline()
# 移除窗口外的记录
pipe.zremrangebyscore(key, 0, window_start)
# 统计当前窗口内的请求数
pipe.zcard(key)
# 添加当前请求
pipe.zadd(key, {str(current): current})
# 设置过期时间
pipe.expire(key, period)
results = pipe.execute()
count = results[1]
return count < limit
# 测试
for i in range(15):
if rate_limit('api:user:1', limit=10, period=60):
print(f'请求{i}：允许')
else:
print(f'请求{i}：拒绝')
案例4：消息队列
PYTHON
import redis
import json
r = redis.Redis(decode_responses=True)
class MessageQueue:
def __init__(self, name):
self.name = f"queue:{name}"
def push(self, message):
# 入队
r.rpush(self.name, json.dumps(message))
def pop(self, timeout=0):
# 出队（阻塞）
result = r.blpop(self.name, timeout=timeout)
if result:
return json.loads(result[1])
return None
def size(self):
return r.llen(self.name)
# 生产者
queue = MessageQueue('tasks')
queue.push({'task': 'send_email', 'to': 'user@example.com'})
queue.push({'task': 'generate_report', 'report_id': 123})
# 消费者
while True:
task = queue.pop(timeout=5)
if task:
print(f"处理任务：{task}")
else:
break
错误处理
PYTHON
import redis
from redis.exceptions import RedisError, ConnectionError
try:
r = redis.Redis(host='localhost', port=6379)
r.set('key', 'value')
except ConnectionError:
print('连接失败')
except RedisError as e:
print(f'Redis错误：{e}')
❓ 常见问题
Q redis-py是线程安全的吗？
A Redis实例是线程安全的，可以在多线程中使用。但建议使用连接池。
Q 如何处理连接断开？
A redis-py会自动重连。可以设置retry_on_timeout=True。
Q decode_responses=True有什么作用？
A 自动将bytes解码为str，避免手动解码。
Q 如何选择redis-py还是aioredis？
A 同步代码用redis-py，异步代码用aioredis（或redis-py 4.2+的异步支持）。
Q 连接池的最大连接数如何设置？
A 根据并发量设置，通常设置为并发线程/进程数的2-3倍。
📖 小节
- redis-py是Python最常用的Redis客户端
- 支持所有Redis命令：字符串、哈希、列表、集合、有序集合
- 连接池复用连接，提高性能
- 管道批量操作，减少网络往返
- 发布订阅实现消息推送
- 实战案例：缓存装饰器、分布式锁、限流器、消息队列
📝 作业
- 基本操作： 使用redis-py实现字符串、哈希、列表、集合的基本操作
- 连接池： 创建连接池，测试连接复用
- 管道： 使用管道批量设置1000个键，对比性能
- 实战： 实现一个简单的缓存装饰器或分布式锁
下一课
下一课我们将学习 Java 使用 Redis，学习Java操作Redis。