管道示例

此示例快速展示如何在 redis-py 中使用管道。

检查 Redis 是否正在运行

import redis

r = redis.Redis(decode_responses=True)
r.ping()

True

简单示例

创建管道实例

pipe = r.pipeline()

将命令添加到管道中

pipe.set("a", "a value")
pipe.set("b", "b value")

pipe.get("a")

Pipeline<ConnectionPool<Connection<host=localhost,port=6379,db=0>>>

执行管道

pipe.execute()

[True, True, 'a value']

三个命令的响应存储在一个列表中。在上面的示例中，前两个布尔值表示 set 命令执行成功，列表的最后一个元素是 get("a") 命令的结果。

链式调用

可以通过将操作链接起来，在一行代码中获得与上面相同的结果。

pipe = r.pipeline()
pipe.set("a", "a value").set("b", "b value").get("a").execute()

[True, True, 'a value']

性能比较

使用管道可以提高性能，更多信息请参见 Redis 关于管道的文档。这是一个关于基本命令和管道命令性能的简单比较测试（我们只是简单地递增一个值，并测量两种方法所花费的时间）。

from datetime import datetime

incr_value = 100000

没有管道

r.set("incr_key", "0")

start = datetime.now()

for _ in range(incr_value):
    r.incr("incr_key")
res_without_pipeline = r.get("incr_key")

time_without_pipeline = (datetime.now() - start).total_seconds()

print("Without pipeline")
print("================")
print("Time taken: ", time_without_pipeline)
print("Increment value: ", res_without_pipeline)

Without pipeline
================
Time taken:  21.759733
Increment value:  100000

使用管道

r.set("incr_key", "0")

start = datetime.now()

pipe = r.pipeline()
for _ in range(incr_value):
    pipe.incr("incr_key")
pipe.get("incr_key")
res_with_pipeline = pipe.execute()[-1]

time_with_pipeline = (datetime.now() - start).total_seconds()

print("With pipeline")
print("=============")
print("Time taken: ", time_with_pipeline)
print("Increment value: ", res_with_pipeline)

With pipeline
=============
Time taken:  2.357863
Increment value:  100000

使用管道在更短的时间内提供了相同的结果。