Python Redis pipeline操作

Redis是建立在TCP协议基础上的CS架构，客户端client对redis server采取请求响应的方式交互。

一般来说客户端从提交请求到得到服务器相应，需要传送两个tcp报文。

设想这样的一个场景，你要批量的执行一系列redis命令，例如执行100次get key，这时你要向redis请求100次+获取响应100次。如果能一次性将100个请求提交给redis server，执行完成之后批量的获取相应，只需要向redis请求1次，然后批量执行完命令，一次性结果，性能是不是会好很多呢？

答案是肯定的，节约的时间是客户端client和服务器redis server之间往返网络延迟的时间。这个时间可以用ping命令查看。

网络延迟高：批量执行，性能提升明显

网络延迟低（本机）：批量执行，性能提升不明显

某些客户端（java和python）提供了一种叫做pipeline的编程模式用来解决批量提交请求的方式。

这里我们用python客户端来举例说明一下。

1、pipeline

网络延迟

client与server机器之间网络延迟如下，大约是30ms。

测试用例

分别执行其中的try_pipeline和without_pipeline统计处理时间。

# -*- coding:utf-8 -*-

import redis
import time
from concurrent.futures import ProcessPoolExecutor

r = redis.Redis(host='localhost', port=6379, password='123456')

def try_pipeline():
    start = time.time()
    with r.pipeline(transaction=False) as p:
        p.sadd('seta', 1).sadd('seta', 2).srem('seta', 2).lpush('lista', 1).lrange('lista', 0, -1)
        p.execute()
    print time.time() - start

def without_pipeline():
    start = time.time()
    r.sadd('seta', 1)
    r.sadd('seta', 2)
    r.srem('seta', 2)
    r.lpush('lista', 1)
    r.lrange('lista', 0, -1)
    print time.time() - start

def worker():
    while True:
        try_pipeline()

with ProcessPoolExecutor(max_workers=12) as pool:
    for _ in range(10):
        pool.submit(worker)

# -*- coding:utf-8 -*-

import redis

import time

from concurrent.futures import ProcessPoolExecutor

r = redis.Redis(host='localhost', port=6379, password='123456')

def try_pipeline():

start = time.time()

with r.pipeline(transaction=False) as p:

p.sadd('seta', 1).sadd('seta', 2).srem('seta', 2).lpush('lista', 1).lrange('lista', 0, -1)

p.execute()

print time.time() - start

def without_pipeline():

start = time.time()

r.sadd('seta', 1)

r.sadd('seta', 2)

r.srem('seta', 2)

r.lpush('lista', 1)

r.lrange('lista', 0, -1)

print time.time() - start

def worker():

while True:

try_pipeline()

with ProcessPoolExecutor(max_workers=12) as pool:

for _ in range(10):

pool.submit(worker)

结果分析

try_pipeline平均处理时间：0.04659

without_pipeline平均处理时间：0.16672

我们的批量里有5个操作，在处理时间维度上性能提升了4倍！

网络延迟大约是30ms，不使用批量的情况下，网络上的时间损耗就有0.15s（30ms*5）以上。而pipeline批量操作只进行一次网络往返，所以延迟只有0.03s。可以看到节省的时间基本都是网路延迟。

2、pipeline与transation

pipeline不仅仅用来批量的提交命令，还用来实现事务transation。

这里对redis事务的讨论不会太多，只是给出一个demo。详细的描述你可以参见这篇博客。redis事务

细心的你可能发现了，使用transaction与否不同之处在与创建pipeline实例的时候，transaction是否打开，默认是打开的。

# -*- coding:utf-8 -*-

import redis
from redis import WatchError
from concurrent.futures import ProcessPoolExecutor

r = redis.Redis(host='127.0.0.1', port=6379)

# 减库存函数, 循环直到减库存完成
# 库存充足, 减库存成功, 返回True
# 库存不足, 减库存失败, 返回False
def decr_stock():

    # python中redis事务是通过pipeline的封装实现的
    with r.pipeline() as pipe:
        while True:
            try:
                # watch库存键, multi后如果该key被其他客户端改变, 事务操作会抛出WatchError异常
                pipe.watch('stock:count')
                count = int(pipe.get('stock:count'))
                if count > 0:  # 有库存
                    # 事务开始
                    pipe.multi()
                    pipe.decr('stock:count')
                    # 把命令推送过去
                    # execute返回命令执行结果列表, 这里只有一个decr返回当前值
                    print pipe.execute()[0]
                    return True
                else:
                    return False
            except WatchError, ex:
                # 打印WatchError异常, 观察被watch锁住的情况
                print ex
                pipe.unwatch()

def worker():
    while True:
        # 没有库存就退出
        if not decr_stock():
            break

# 实验开始
# 设置库存为100
r.set("stock:count", 100)

# 多进程模拟多个客户端提交
with ProcessPoolExecutor(max_workers=2) as pool:
    for _ in range(10):
        pool.submit(worker)

# -*- coding:utf-8 -*-

import redis

from redis import WatchError

from concurrent.futures import ProcessPoolExecutor

r = redis.Redis(host='127.0.0.1', port=6379)

# 减库存函数, 循环直到减库存完成

# 库存充足, 减库存成功, 返回True

# 库存不足, 减库存失败, 返回False

def decr_stock():

# python中redis事务是通过pipeline的封装实现的

with r.pipeline() as pipe:

while True:

try:

# watch库存键, multi后如果该key被其他客户端改变, 事务操作会抛出WatchError异常

pipe.watch('stock:count')

count = int(pipe.get('stock:count'))

if count > 0: # 有库存

# 事务开始

pipe.multi()

pipe.decr('stock:count')

# 把命令推送过去

# execute返回命令执行结果列表, 这里只有一个decr返回当前值

print pipe.execute()[0]

return True

else:

return False

except WatchError, ex:

# 打印WatchError异常, 观察被watch锁住的情况

print ex

pipe.unwatch()

def worker():

while True:

# 没有库存就退出

if not decr_stock():

break

# 实验开始

# 设置库存为100

r.set("stock:count", 100)

# 多进程模拟多个客户端提交

with ProcessPoolExecutor(max_workers=2) as pool:

for _ in range(10):

pool.submit(worker)

观察打印

99
98
97
Watched variable changed.
96
95
94
93
Watched variable changed.
92
Watched variable changed.
91
Watched variable changed.
90
Watched variable changed.
89
Watched variable changed.
88
Watched variable changed.
Watched variable changed.
87
86
Watched variable changed.
85
Watched variable changed.
84
Watched variable changed.
Watched variable changed.
83
82
Watched variable changed.
81
Watched variable changed.
Watched variable changed.
80
79
Watched variable changed.
Watched variable changed.
78
77
Watched variable changed.
Watched variable changed.
76
75
Watched variable changed.
Watched variable changed.74

Watched variable changed.
73
72
Watched variable changed.
Watched variable changed.
71
70
Watched variable changed.
69
Watched variable changed.
68
Watched variable changed.
67
Watched variable changed.
66
Watched variable changed.
Watched variable changed.65

64
Watched variable changed.
63
Watched variable changed.
Watched variable changed.
62
Watched variable changed.
61
60
Watched variable changed.
59
Watched variable changed.
Watched variable changed.
58
57
Watched variable changed.
Watched variable changed.
56
Watched variable changed.
55
54
Watched variable changed.
53
Watched variable changed.
52
Watched variable changed.
Watched variable changed.
51
50
Watched variable changed.
49
Watched variable changed.
48
Watched variable changed.
47
Watched variable changed.
Watched variable changed.46

Watched variable changed.
45
Watched variable changed.
44
43
Watched variable changed.
42
Watched variable changed.
Watched variable changed.
41
40
Watched variable changed.
Watched variable changed.
39
Watched variable changed.
38
Watched variable changed.
37
Watched variable changed.36

Watched variable changed.
35
34
Watched variable changed.
33
Watched variable changed.
Watched variable changed.32

Watched variable changed.
31
30
Watched variable changed.
Watched variable changed.
29
Watched variable changed.
28
Watched variable changed.27

26
Watched variable changed.
25
Watched variable changed.
24
Watched variable changed.
23
Watched variable changed.
22Watched variable changed.

Watched variable changed.21

20Watched variable changed.

19
Watched variable changed.
18
Watched variable changed.
17
Watched variable changed.
16
Watched variable changed.
Watched variable changed.
15
Watched variable changed.
14
Watched variable changed.
13
12
Watched variable changed.
Watched variable changed.
11
Watched variable changed.
10
Watched variable changed.9

8
Watched variable changed.
7
Watched variable changed.
Watched variable changed.
6
5
Watched variable changed.
Watched variable changed.
4
Watched variable changed.
3
2
Watched variable changed.
1
Watched variable changed.
0
Watched variable changed.

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

Watched variable changed.

Watched variable changed.74

Watched variable changed.

Watched variable changed.65

Watched variable changed.

Watched variable changed.46

Watched variable changed.

Watched variable changed.36

Watched variable changed.

Watched variable changed.32

Watched variable changed.

Watched variable changed.27

Watched variable changed.

22Watched variable changed.

Watched variable changed.21

20Watched variable changed.

Watched variable changed.

Watched variable changed.9

Watched variable changed.

织梦先生

一个从零开始并且为了自己心底的理想不断奋斗中的人

Python Redis pipeline操作