cas全稱是compare and set，是一種典型的事務(wù)操作，本文會介紹三種redis實現(xiàn)cas事務(wù)的方法，并會解決下面的虛擬問題：
維護一個值，如果這個值小于當前時間，則設(shè)置為當前時間；如果這個值大于當前時間，則設(shè)置為當前時間+30。簡單的單線程環(huán)境下代碼如下:
# 初始化
r = redis.Redis()
if not r.exists("key_test"):
r.set("key_test", 0)
def inc():
count = int(r.get('key_test')) + 30 #1
# 如果值比當前時間小，則設(shè)置為當前時間
count = max(count, int(time.time())) #2
r.set('key_test', count) #3
return count
很簡單的一段代碼，在單線程環(huán)境下可以跑的很歡，但顯然，是無法移植到多線程或者是多進程環(huán)境的（進程A和B同時運行到#1，獲取了相同的count值，然后運行#2#3，會導(dǎo)致count值總共只增加了30）。而為了能在多進程環(huán)境下運行，我們需要引入一些其他的東西。
py-redis本身自帶的事務(wù)操作
redis有這么幾個和事務(wù)相關(guān)的命令，multi,exec,watch。通過這幾個命令，可以實現(xiàn)‘將多個命令打包，然后一次性、按順序執(zhí)行，且不會被終端’。事務(wù)會從MULTI開始，執(zhí)行EXEC后觸發(fā)事件。另外，我們還需要WATCH，watch可以監(jiān)視任意數(shù)量的鍵，當在調(diào)用EXEC執(zhí)行事務(wù)時，如果任意一個鍵被修改了，整個事務(wù)不會執(zhí)行。
下邊是使用redis本身的事務(wù)解決cas問題的代碼。
class CasNormal(object):
def __init__(self, host, key):
self.r = redis.Redis(host)
self.key = key
if not self.r.exists(self.key):
self.r.set(self.key, 0)
def inc(self):
with self.r.pipeline() as pipe:
while True:
try:
#監(jiān)視一個key，如果在執(zhí)行期間被修改了，會拋出WatchError
pipe.watch(self.key)
next_count = 30 + int(pipe.get(self.key))
pipe.multi()
if next_count < int(time.time()):
next_count = int(time.time())
pipe.set(self.key, next_count)
pipe.execute()
return next_count
except WatchError:
continue
finally:
pipe.reset()
代碼也不復(fù)雜，引入了之前說到的multi,exec,watch，如果對事務(wù)操作比較熟悉的同學(xué)，可以很容易看出來，這是一個樂觀鎖的操作（咱們假設(shè)沒人競爭來著，每次去拿數(shù)據(jù)的時候都不會上鎖，真有人來改了再說。）樂觀鎖在高并發(fā)的情況下會顯得很無力，文末的性能對比會顯示這個問題。
使用基于redis的悲觀鎖
悲觀鎖，就是很悲觀的鎖，每次拿數(shù)據(jù)都會假設(shè)別人也要拿，先給鎖起來，用完再把鎖釋放掉。redis本身沒有實現(xiàn)悲觀鎖，但我們可以先用redis實現(xiàn)一個悲觀鎖。
此處應(yīng)該有個推倒出redis悲觀鎖的過程，不過太麻煩了...直接丟個鏈接吧...
https://gist.github.com/gaoconghui/61e878c725952c134a1193d560df7434
ok，咱們現(xiàn)在有悲觀鎖了，做起事來也有底氣了，根據(jù)上邊的代碼，咱們只要加上@ synchronized注釋就能保證同一時間只有一個進程在執(zhí)行。下邊是基于悲觀鎖的解決方案。
lock_conn = redis.Redis("localhost")
class CasLock(object):
def __init__(self, host, key):
self.r = redis.Redis(host)
self.key = key
if not self.r.exists(self.key):
self.r.set(self.key, 0)
@synchronized(lock_conn, "lock", 10)
def inc(self):
next_count = 30 + int(self.r.get(self.key))
if next_count < int(time.time()):
next_count = int(time.time())
self.r.set(self.key, next_count)
return next_count
代碼看上去少多了（因為引入了synchronized...）
基于lua腳本實現(xiàn)
上邊兩種方法都是用鎖來實現(xiàn)的，鎖的實現(xiàn)總會出現(xiàn)競爭的問題，區(qū)別無非是出現(xiàn)競爭了咋辦的問題。使用redis lua腳本的實現(xiàn)，可以直接把這個cas操作當成一個<b>原子操作</b>。
我們知道，redis本身的一系列操作，都是原子操作，且redis會按順序執(zhí)行所有收到的命令。先看代碼
class CasLua(object):
def __init__(self, host, key):
self.r = redis.Redis(host)
self.key = key
if not self.r.exists(self.key):
self.r.set(self.key, 0)
self._lua = self.r.register_script("""
local next_count = redis.call('get',KEYS[1]) + ARGV[1]
ARGV[2] = tonumber(ARGV[2])
if next_count < ARGV[2] then
next_count = ARGV[2]
end
redis.call('set',KEYS[1],next_count)
return tostring(next_count)
""")
def inc(self):
return int(self._lua([self.key], [30, int(time.time())]))
這里先注冊了這個腳本，后邊可以直接去使用他。關(guān)于redis lua腳本的文章有不少，感興趣的可以去搜搜看，這邊就不贅述了。
性能對比
這邊的測試只是一個非常簡單的測試（不過還是能看出效果來的），測試換機就是自己的開發(fā)機，數(shù)字看個大小就行了。
分別測了三種操作在單線程，五個線程，十個線程，五十個線程情況下，進行1000次操作各自的表現(xiàn)，時間如下
optimistic Lock pessimistic lock lua
1thread 0.43 0.71 0.35
5thread 5.80 3.10 0.62
10thread 17.80 5.60 1.30
50thread 245.00 29.60 6.50
依次是redis本身事務(wù)實現(xiàn)的樂觀鎖，基于redis實現(xiàn)的悲觀鎖以及l(fā)ua實現(xiàn)。
在比較悲觀鎖和樂觀鎖之前，需要先說明一點，這邊的測試對樂觀鎖不是很公平，樂觀鎖本身就是假設(shè)不會有很多的并發(fā)的。在單線程情況下，悲觀鎖要差一些。單線程下，不存在競爭關(guān)系，悲觀鎖耗時長僅因為是多了一次redis的網(wǎng)絡(luò)交互。隨著線程的增加，悲觀鎖的性能逐漸變好，畢竟悲觀鎖本身就是為了解決這種高并發(fā)高競爭的環(huán)境而誕生的。在50線程的時候，樂觀鎖的實現(xiàn)單次操作的時間要0.245秒，非?？植?，如果是生產(chǎn)環(huán)境，幾乎都不能用了。
至于lua的性能，快的不可思議，幾乎就是線性增加。（50線程的情況下，平均的1000次完成時間是6.5s，換言之，6.5秒內(nèi)執(zhí)行了50 * 1000次cas操作）。
以上測試都是本地redis，本地測試，如果redis是遠端的，網(wǎng)絡(luò)交互時間會增加，lua優(yōu)勢會更加明顯。
以上。
