<div id="article_content" class="article_content clearfix csdn-tracking-statistics" data-pid="blog" data-mod="popu_307" data-dsm="post">
<div class="htmledit_views">
Redis是建立在TCP协议基础上的CS架构，客户端client对redis server采取请求响应的方式交互。

redis 乐观锁：也可理解为版本号比较机制，主要是说在读取数据逇时候同时读取其版本号，然后在写入的时候，进行版本号比较，如果一致，则表明此数据在监听期间未被改变，可以写入，如果不一致说明此数据被修改过，不能写入，否则会导致数据不一致的问题。

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

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

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

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

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

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

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

<h1 style="margin-left: 10px;">1、pipeline
网络延迟

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

测试用例

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


<span style="color: #0000ff;">import<span style="color: #000000;"> redis

<span style="color: #0000ff;">import<span style="color: #000000;"> time

<span style="color: #0000ff;">from concurrent.futures <span style="color: #0000ff;">import<span style="color: #000000;"> ProcesspoolExecutor
r = redis.Redis(host=<span style="color: #800000;">'<span style="color: #800000;">10.93.84.53<span style="color: #800000;">',port=6379,password=<span style="color: #800000;">'<span style="color: #800000;">bigdata123<span style="color: #800000;">'<span style="color: #000000;">)
<span style="color: #0000ff;">def<span style="color: #000000;"> try_pipeline():

start =<span style="color: #000000;"> time.time()

with r.pipeline(transaction=<span style="color: #000000;">False) as p:

p.sadd(<span style="color: #800000;">'<span style="color: #800000;">seta<span style="color: #800000;">',1).sadd(<span style="color: #800000;">'<span style="color: #800000;">seta<span style="color: #800000;">',2).srem(<span style="color: #800000;">'<span style="color: #800000;">seta<span style="color: #800000;">',2).lpush(<span style="color: #800000;">'<span style="color: #800000;">lista<span style="color: #800000;">',1).lrange(<span style="color: #800000;">'<span style="color: #800000;">lista<span style="color: #800000;">',-1<span style="color: #000000;">)

p.execute()

<span style="color: #0000ff;">print time.time() -<span style="color: #000000;"> start
<span style="color: #0000ff;">def<span style="color: #000000;"> without_pipeline():

start =<span style="color: #000000;"> time.time()

r.sadd(<span style="color: #800000;">'<span style="color: #800000;">seta<span style="color: #800000;">',1<span style="color: #000000;">)

r.sadd(<span style="color: #800000;">'<span style="color: #800000;">seta<span style="color: #800000;">',2<span style="color: #000000;">)

r.srem(<span style="color: #800000;">'<span style="color: #800000;">seta<span style="color: #800000;">',2<span style="color: #000000;">)

r.lpush(<span style="color: #800000;">'<span style="color: #800000;">lista<span style="color: #800000;">',1<span style="color: #000000;">)

r.lrange(<span style="color: #800000;">'<span style="color: #800000;">lista<span style="color: #800000;">',-1<span style="color: #000000;">)

<span style="color: #0000ff;">print time.time() -<span style="color: #000000;"> start
<span style="color: #0000ff;">def<span style="color: #000000;"> worker():

<span style="color: #0000ff;">while<span style="color: #000000;"> True:

try_pipeline()
with ProcesspoolExecutor(maxworkers=12<span style="color: #000000;">) as pool:

<span style="color: #0000ff;">for  <span style="color: #0000ff;">in range(10<span style="color: #000000;">):

pool.submit(worker)

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():

</span><span style="color: #008000;"&gt;#</span><span style="color: #008000;"&gt; python中redis事务是通过pipeline的封装实现的</span>

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(maxworkers=2) as pool:
for in range(10):
pool.submit(worker)

Python Redis pipeline操作和Redis乐观锁保持数据一致性

<span style="color: #008000;"> 库存充足,减库存成功,返回True<span style="color: #008000;">

<span style="color: #008000;"> 库存不足,减库存失败,返回False

<span style="color: #008000;"> 设置库存为100

相关推荐