Redis 管道技术
管道（Pipeline）可以批量发送命令，减少网络往返次数，大幅提升性能。
什么是管道？
传统方式
每条命令都需要一次网络往返：
客户端 Redis服务器
| |
|----- SET key1 -------->|
|<----- OK --------------|
| |
|----- SET key2 -------->|
|<----- OK --------------|
| |
|----- SET key3 -------->|
|<----- OK --------------|
3条命令 = 6次网络传输（3次请求 + 3次响应）
管道方式
多条命令一次性发送：
客户端 Redis服务器
| |
|----- SET key1 -------->|
|----- SET key2 -------->|
|----- SET key3 -------->|
| |
|<----- OK --------------|
|<----- OK --------------|
|<----- OK --------------|
3条命令 = 2次网络传输（1次批量请求 + 1次批量响应）
💡 性能提升： 管道可以减少网络往返，大幅提升性能，尤其是网络延迟较高时。
管道原理
网络延迟影响
假设网络往返延迟为1ms：
传统方式：
- 100条命令 = 100次往返 = 100ms网络延迟
- 加上Redis处理时间（假设0.1ms/命令）= 10ms
- 总时间 = 100ms + 10ms = 110ms
管道方式：
- 100条命令 = 1次往返 = 1ms网络延迟
- 加上Redis处理时间 = 10ms
- 总时间 = 1ms + 10ms = 11ms
性能提升：110ms → 11ms，提升10倍！
管道特点
- 批量发送：多个命令一次性发送
- 顺序执行：Redis按顺序执行命令
- 批量返回：所有结果一次性返回
- 非原子性：不是事务，命令可能被其他客户端打断
⚠️ 注意： 管道不是事务，不保证原子性。如需原子性，使用MULTI/EXEC。
使用管道
redis-cli管道模式
BASH
# 使用redis-cli的管道模式
echo -e "SET key1 value1\nSET key2 value2\nSET key3 value3" | redis-cli
# 或从文件读取
cat commands.txt | redis-cli
Python使用管道
PYTHON
import redis
r = redis.Redis(host='localhost', port=6379)
# 创建管道
pipe = r.pipeline()
# 添加命令到管道
pipe.set('key1', 'value1')
pipe.set('key2', 'value2')
pipe.set('key3', 'value3')
# 执行管道
results = pipe.execute()
print(results) # [True, True, True]
Java使用管道（Jedis）
JAVA
Jedis jedis = new Jedis("localhost", 6379);
Pipeline pipeline = jedis.pipelined();
pipeline.set("key1", "value1");
pipeline.set("key2", "value2");
pipeline.set("key3", "value3");
List<Object> results = pipeline.syncAndReturnAll();
Node.js使用管道（ioredis）
JAVASCRIPT
const Redis = require('ioredis');
const redis = new Redis();
const pipeline = redis.pipeline();
pipeline.set('key1', 'value1');
pipeline.set('key2', 'value2');
pipeline.set('key3', 'value3');
const results = await pipeline.exec();
管道与事务
管道 + 事务
管道可以与事务结合使用：
PYTHON
import redis
r = redis.Redis()
# 创建管道，开启事务
pipe = r.pipeline(transaction=True)
pipe.set('key1', 'value1')
pipe.set('key2', 'value2')
pipe.incr('counter')
# 执行事务
results = pipe.execute()
管道 vs 事务
| 对比项 | 管道 | 事务 |
|---|---|---|
| 原子性 | ❌ 无 | ✅ 有 |
| 隔离性 | ❌ 无 | ✅ 有 |
| 性能提升 | ✅ 显著 | ✅ 显著 |
| 批量发送 | ✅ 是 | ✅ 是 |
| 适用场景 | 批量操作、提升性能 | 需要原子性的批量操作 |
💡 选择： 不需要原子性时用管道，需要原子性时用事务或管道+事务。
管道应用场景
场景1：批量设置
PYTHON
import redis
r = redis.Redis()
pipe = r.pipeline()
# 批量设置1000个键
for i in range(1000):
pipe.set(f'key:{i}', f'value:{i}')
results = pipe.execute()
print(f'设置完成：{len(results)}个键')
场景2：批量获取
PYTHON
import redis
r = redis.Redis()
pipe = r.pipeline()
# 批量获取1000个键
keys = [f'key:{i}' for i in range(1000)]
for key in keys:
pipe.get(key)
values = pipe.execute()
print(f'获取完成：{len(values)}个值')
场景3：批量删除
PYTHON
import redis
r = redis.Redis()
pipe = r.pipeline()
# 批量删除
keys = ['key1', 'key2', 'key3', 'key4', 'key5']
for key in keys:
pipe.delete(key)
results = pipe.execute()
print(f'删除完成：{sum(results)}个键')
场景4：数据导入
PYTHON
import redis
import json
r = redis.Redis()
pipe = r.pipeline()
# 从JSON文件导入数据
with open('data.json', 'r') as f:
data = json.load(f)
for item in data:
key = f"user:{item['id']}"
value = json.dumps(item)
pipe.set(key, value)
pipe.execute()
print('导入完成')
场景5：计数器批量更新
PYTHON
import redis
r = redis.Redis()
pipe = r.pipeline()
# 批量更新计数器
counters = {
'article:1:views': 10,
'article:2:views': 20,
'article:3:views': 30,
}
for key, increment in counters.items():
pipe.incrby(key, increment)
pipe.execute()
print('计数器更新完成')
管道性能对比
测试代码
PYTHON
import redis
import time
r = redis.Redis()
# 测试数据
n = 10000
# 方式1：普通方式
start = time.time()
for i in range(n):
r.set(f'key:{i}', f'value:{i}')
end = time.time()
print(f'普通方式：{end - start:.2f}秒')
# 方式2：管道方式
start = time.time()
pipe = r.pipeline()
for i in range(n):
pipe.set(f'key:{i}', f'value:{i}')
pipe.execute()
end = time.time()
print(f'管道方式：{end - start:.2f}秒')
测试结果示例
普通方式：5.23秒
管道方式：0.51秒
性能提升：约10倍
管道最佳实践
1. 合理设置批量大小
PYTHON
# ❌ 一次性发送太多命令
pipe = r.pipeline()
for i in range(100000):
pipe.set(f'key:{i}', f'value:{i}')
pipe.execute() # 可能占用大量内存
# ✅ 分批执行
batch_size = 1000
for batch in range(0, 100000, batch_size):
pipe = r.pipeline()
for i in range(batch, batch + batch_size):
pipe.set(f'key:{i}', f'value:{i}')
pipe.execute()
2. 错误处理
PYTHON
import redis
r = redis.Redis()
pipe = r.pipeline()
pipe.set('key1', 'value1')
pipe.incr('key1') # 这会失败（key1不是数字）
pipe.set('key2', 'value2')
try:
results = pipe.execute()
for i, result in enumerate(results):
if isinstance(result, Exception):
print(f'命令{i}失败：{result}')
except Exception as e:
print(f'管道执行失败：{e}')
3. 使用事务保证原子性
PYTHON
import redis
r = redis.Redis()
# 需要原子性时，使用事务
pipe = r.pipeline(transaction=True)
pipe.set('account:a', '100')
pipe.set('account:b', '50')
try:
results = pipe.execute()
print('事务执行成功')
except Exception as e:
print(f'事务执行失败：{e}')
4. 监控管道性能
PYTHON
import redis
import time
r = redis.Redis()
start = time.time()
pipe = r.pipeline()
for i in range(10000):
pipe.set(f'key:{i}', f'value:{i}')
results = pipe.execute()
end = time.time()
print(f'执行时间：{end - start:.2f}秒')
print(f'QPS：{10000 / (end - start):.0f}')
管道限制
1. 内存占用
PYTHON
# 管道会缓存所有命令和结果
# 批量太大时占用大量内存
# 解决方案：分批执行
2. 非原子性
PYTHON
# 管道中的命令可能被其他客户端打断
# 如需原子性，使用事务
3. 无法使用中间结果
PYTHON
# 管道中无法使用前一条命令的结果
# ❌ 错误示例
pipe = r.pipeline()
pipe.incr('counter')
# 无法获取incr的结果用于后续命令
pipe.set('result', ???) # 无法使用incr结果
pipe.execute()
# ✅ 解决方案：使用Lua脚本
❓ 常见问题
Q 管道和事务的区别？
A 管道减少网络往返，不保证原子性。事务保证原子性，也减少网络往返。
Q 管道能提升多少性能？
A 取决于网络延迟。网络延迟越高，提升越明显。通常提升5-10倍。
Q 管道批量大小如何设置？
A 建议每批100-1000条命令。太大占用内存，太小性能提升不明显。
Q 管道中的命令失败会怎样？
A 某条命令失败不影响其他命令执行。需要检查返回结果。
Q 何时使用管道？
A 批量操作、数据导入导出、需要提升性能的场景。
📖 小节
- 管道批量发送命令，减少网络往返
- 性能提升显著，尤其是网络延迟较高时
- 管道不保证原子性，需要原子性时用事务
- 应用场景：批量设置、批量获取、数据导入
- 最佳实践：合理设置批量大小、错误处理、分批执行
- 管道限制：内存占用、非原子性、无法使用中间结果
📝 作业
- 基本管道： 使用管道批量设置100个键，对比普通方式的性能
- 批量操作： 使用管道实现批量获取、批量删除
- 性能对比： 测试不同批量大小（10、100、1000）的性能差异
- 管道+事务： 使用管道+事务实现原子性的批量操作
下一课
下一课我们将学习 Python 使用 Redis，学习Python操作Redis。