python redis事务_Redis事务详解

Redis的基本事务(basic transaction)需要用到MULTI命令和EXEC命令，这种事务可以让一个客户端在不被其他客户端打断的情况下执行多个命令。被MULTI命令和EXEC命令包围的所有命令会一个接一个地执行，直到所有命令都执行完毕为止。当一个事务执行完毕之后，Redis才会处理其他客户端的命令。

当Redis从一个客户端那里接收到MULTI命令时，Redis会将这个客户端之后发送的所有命令都放入到一个队列里面，直到这个客户端发送EXEC命令为止，然后Redis就会在不被打断的情况下，一个接一个地执行存储在队列里面的命令。从语义上来说，Redis事务在Python客户端上面是由流水线(pipeline)实现的：对连接对象调用pipeline()方法将创建一个事务，在一切正常的情况下，客户端会自动地使用MULTI和EXEC包裹起用户输入的多个命令。为了减少Redis与客户端之间的通信往返次数，提升执行多个命令时的性能，Python的Redis客户端会存储起事务包含的多个命令，然后在事务执行时一次性地将所有命令都发送给Redis。

在Python中使用事务来处理命令的并行执行问题：

def trans():

pipeline = conn.pipeline() # 创建事务型流水线对象

pipeline.incr('trans:') # 把针对'trans:'计数器的自增操作放入队列

time.sleep(.1) # 等待100ms

pipeline.incr('trans:', -1) # 把针对'trans:'计数器的自减操作放入队列

print pipeline.execute()[0] # 执行被事务包裹的命令，并打印自增操作的执行结果

if 1:

for i in xrange(3):

# 启动3个线程来执行被事务包裹的自增、休眠和自减3个操作

threading.Thread(target=trans).start()

time.sleep(.5) # 等待500ms,让操作有足够的时间完成

# 打印结果：

1

1

1

Redis要在接收到EXEC命令之后，才会执行哪些位于MULTI和EXEC之间的入队命令。

上述这种简单的事务在EXEC命令被调用之前不会执行任何实际操作，所以用户将没办法根据读取到的数据来做决定。这种方式无法以一致的形式读取数据将导致某一类型的问题变得难以解决，除此之外，因为在多个事务同时处理同一个对象时通常需要用到二阶提交(two-phase commit), 所以如果事务不能以一致的形式读取数据，那么二阶提交将无法实现，从而导致一些原本可以成功执行的事务执行失败。

延迟执行事务有助于提升性能

因为Redis在执行事务的过程中，会延迟执行已入队的命令直到客户端发送EXEC命令为止。包括python客户端在内的很多Redis客户端都会等到事务包含的所有命令都出现了之后，才一次性地将MULTI命令、要在事务中执行的一系列命令，以及EXEC命令全部发送给Redis，然后等待直到接收到所有命令的回复为止。这种“一次性发送多个命令，然后等待所有回复出现”的做法通常被称为流水线(pipeline),它可以通过减少客户端与Redis服务器之间的网络通信次数来提升Redis在执行多个命令时的性能。

在用户使用WATCH命令对键进行监视之后，直到用户执行EXEC命令的这段时间，如果有其他客户端抢先对任何被监视的键进行了替换、更新或删除等操作，那么当用户尝试执行EXEC命令的时候，事务将失败并返回一个错误(之后选择重试事务或者放弃事务)。

UNWATCH命令可以在WATCH命令执行之后、MULTI命令执行之前对连接进行重置(reset);同样地，DISCARD命令也可以在MULTI命令执行之后、EXEC命令执行之前对连接进行重置。这也就是说，用户在使用WATCH监视一个或多个键，接着使用MULTI开始一个新的事务，并将多个命令入队到事务队列之后，仍然可以通过发送DISCARD命令来取消WATCH命令并清空所有已入队命令。

将商品放到市场上销售：

def list_item(conn, itemid, sellerid, price):

inventory = "inventory:%s"%sellerid # 商家包裹

item = "%s.%s"%(itemid, sellerid)

end = time.time() + 5

pipe = conn.pipeline()

while time.time() < end:

try:

pipe.watch(inventory) # 监视商家包裹发生的变化

if not pipe.sismember(inventory, itemid): # 检查商家是否仍然持有将要被销售的商品

pipe.unwatch()

return None

pipe.multi()

pipe.zadd("market:", item, price) # 将出售的商品添加到买卖市场

pipe.srem(inventory, itemid)

pipe.execute() # 执行execute没有引发WatchError异常，说明事务执行成功，并且对包裹键的监视也已经结束

return True

except redis.exceptions.WatchError: # 商家的包裹已经发生变化，重试

pass

return False

购买商品：

def purchase_item(conn, buyerid, itemid, sellerid, lprice):

buyer = "users:%s"%buyerid

seller = "users:%s"%sellerid

item = "%s.%s"%(itemid, sellerid)

inventory = "inventory:%s"%buyerid

end = time.time() + 10

pipe = conn.pipeline()

while time.time() < end:

try:

pipe.watch("market:", buyer) # 对商品买卖市场以及买家的个人信息进行监视

# 检查购买的商品价格是否发生变化，以及卖家是否有足够的钱购买

price = pipe.zscore("market:", item)

funds = int(pipe.hget(buyer, "funds"))

if price != lprice or price > funds:

pipe.unwatch()

return None

pipe.multi()

pipe.hincrby(seller, "funds", int(price))

pipe.hincrby(buyer, "funds", int(-price))

pipe.sadd(inventory, itemid)

pipe.zrem("market:", item)

pipe.execute()

return True

except redis.exceptions.WatchError:

pass

return False

加锁有可能造成长时间的等待，所以Redis为了尽可能地减少客户端的等待时间，并不会在执行WATCH命令时对数据进行加锁。相反，Redis只会在数据已经被其他客户端抢先修改了的情况下，通知执行WATCH命令的客户端，这种做法称为乐观锁(optimistic locking),而关系型数据库实际执行的加锁操作则被称为悲观锁(pessimistic locking)。