單線程與隔離性 Redis是使用單線程的方式來執行事務的，事務以串列的方式運行，也就是說Redis中單個命令的執行和事務的執行都是線程安全的，不會相互影響，具有隔離性。 在多線程編程中，對於共用資源的訪問要十分的小心： 在不加鎖的情況下，num是不能保持為1的。 而在Redis中，併發執行單個命令具 ...
單線程與隔離性
Redis是使用單線程的方式來執行事務的，事務以串列的方式運行，也就是說Redis中單個命令的執行和事務的執行都是線程安全的，不會相互影響，具有隔離性。
在多線程編程中，對於共用資源的訪問要十分的小心：
import threading
num = 1
lock = threading.Lock()
def change_num():
global num
for i in xrange(100000):
#lock.acquire()
num += 5
num -= 5
#lock.release()
if __name__ == '__main__':
pool = [threading.Thread(target=change_num) for i in xrange(5)]
for t in pool:
t.start()
for t in pool:
t.join()
print num
在不加鎖的情況下，num是不能保持為1的。
而在Redis中，併發執行單個命令具有很好的隔離性：
import redis
conn = redis.StrictRedis(host="localhost", port=6379, db=1)
conn.set('num', 1)
def change_num(conn):
for i in xrange(100000):
┆ conn.incr('num', 5)
┆ conn.decr('num', 5)
if __name__ == '__main__':
conn_pool = [redis.StrictRedis(host="localhost", port=6379, db=1)
for i in xrange(5)]
t_pool = []
for conn in conn_pool:
t = threading.Thread(target=change_num, args=(conn,))
t_pool.append(t)
for t in t_pool:
t.start()
for t in t_pool:
t.join()
print conn.get('num')
模擬的5個客戶端同時對Redis中的num值進行操作，num最終結果會保持為1：
1 real 0m46.463s user 0m28.748s sys 0m6.276s
利用Redis中單個操作和事務的原子性可以做很多事情，最簡單的就是做全局計數器了。
比如在簡訊驗證碼業務中，要限制一個用戶在一分鐘內只能發送一次，如果使用關係型資料庫，需要為每個手機號記錄上次發送簡訊的時間，當用戶請求驗證碼時，取出與當前時間進行對比。
這一情況下，當用戶短時間點擊多次時，不僅增加了資料庫壓力，而且還會出現同時查詢均符合條件但資料庫更新簡訊發送時間較慢的問題，就會重覆發送簡訊了。
在Redis中解決這一問題就很簡單，只需要用手機號作為key創建一個生存期限為一分鐘的數值即可。key不存在時能發送簡訊，存在時則不能發送簡訊：
def can_send(phone):
key = "message:" + str(phone)
if conn.set(key, 0, nx=True, ex=60):
┆ return True
else:
┆ return False
至於一些不可名的30分鐘內限制訪問或者下載5次的功能，將用戶ip作為key，值設為次數上限，過期時間設為限制時間，每次用戶訪問時自減即可：
def can_download(ip):
key = "ip:" + str(ip)
conn.set(key, 5, nx=True, ex=600)
if conn.decr(key) >= 0:
┆ return True
else:
┆ return False
Redis基本事務與樂觀鎖
雖然Redis單個命令具有原子性，但當多個命令並行執行的時候，會有更多的問題。
比如舉一個轉賬的例子，將用戶A的錢轉給用戶B，那麼用戶A的賬戶減少需要與B賬戶的增多同時進行：
import threading
import time
import redis
conn = redis.StrictRedis(host="localhost", port=6379, db=1)
conn.mset(a_num=10, b_num=10)
def a_to_b():
if int(conn.get('a_num')) >= 10:
conn.decr('a_num', 10)
time.sleep(.1)
conn.incr('b_num', 10)
print conn.mget('a_num', "b_num")
def b_to_a():
if int(conn.get('b_num')) >= 10:
conn.decr('b_num', 10)
time.sleep(.1)
conn.incr('a_num', 10)
print conn.mget('a_num', "b_num")
if __name__ == '__main__':
pool = [threading.Thread(target=a_to_b) for i in xrange(3)]
for t in pool:
t.start()
pool = [threading.Thread(target=b_to_a) for i in xrange(3)]
for t in pool:
t.start()
運行結果：
['0', '10'] ['0', '10'] ['0', '0'] ['0', '0'] ['0', '10'] ['10', '10']
出現了賬戶總額變少的情況。雖然是人為的為自增自減命令之間添加了100ms延遲，但在實際併發很高的情況中是很可能出現的，兩個命令執行期間執行了其它的語句。
那麼現在要保證的是兩個增減命令執行期間不受其它命令的干擾，Redis的事務可以達到這一目的。
Redis中，被MULTI命令和EXEC命令包圍的所有命令會一個接一個的執行，直到所有命令都執行完畢為止。一個事務完畢後，Redis才會去處理其它的命令。也就是說，Redis事務是具有原子性的。
python中可以用pipeline來創建事務：
def a_to_b():
if int(conn.get('a_num')) >= 10:
┆ pipeline = conn.pipeline()
┆ pipeline.decr('a_num', 10)
┆ time.sleep(.1)
┆ pipeline.incr('b_num', 10)
┆ pipeline.execute()
print conn.mget('a_num', "b_num")
def b_to_a():
if int(conn.get('b_num')) >= 10:
┆ pipeline = conn.pipeline()
┆ pipeline.decr('b_num', 10)
┆ time.sleep(.1)
┆ pipeline.incr('a_num', 10)
┆ pipeline.execute()
print conn.mget('a_num', "b_num")
結果：
['0', '20'] ['10', '10'] ['-10', '30'] ['-10', '30'] ['0', '20'] ['10', '10']
可以看到，兩條語句確實一起執行了，賬戶總額不會變，但出現了負值的情況。這是因為事務在exec命令被調用之前是不會執行的，所以用讀取的數據做判斷與事務執行之間就有了時間差，期間實際數據發生了變化。
為了保持數據的一致性，我們還需要用到一個事務命令WATCH。WATCH可以對一個鍵進行監視，監視後到EXEC命令執行之前，如果被監視的鍵值發生了變化（替換，更新，刪除等），EXEC命令會返回一個錯誤，而不會真正的執行：
>>> pipeline.watch('a_num')
True
>>> pipeline.multi()
>>> pipeline.incr('a_num',10)
StrictPipeline<ConnectionPool<Connection<host=localhost,port=6379,db=1>>>
>>> pipeline.execute()
[20]
>>> pipeline.watch('a_num')
True
>>> pipeline.incr('a_num',10) #監視期間改變被監視鍵的值
30
>>> pipeline.multi()
>>> pipeline.incr('a_num',10)
StrictPipeline<ConnectionPool<Connection<host=localhost,port=6379,db=1>>>
>>> pipeline.execute()
raise WatchError("Watched variable changed.")
redis.exceptions.WatchError: Watched variable changed.
現在為代碼加上watch：
def a_to_b():
pipeline = conn.pipeline()
try:
┆ pipeline.watch('a_num')
┆ if int(pipeline.get('a_num')) < 10:
┆ ┆ pipeline.unwatch()
┆ ┆ return
┆ pipeline.multi()
┆ pipeline.decr('a_num', 10)
┆ pipeline.incr('b_num', 10)
┆ pipeline.execute()
except redis.exceptions.WatchError:
┆ pass
print conn.mget('a_num', "b_num")
def b_to_a():
pipeline = conn.pipeline()
try:
┆ pipeline.watch('b_num')
┆ if int(pipeline.get('b_num')) < 10:
┆ ┆ pipeline.unwatch()
┆ ┆ return
┆ pipeline.multi()
┆ pipeline.decr('b_num', 10)
┆ pipeline.incr('a_num', 10)
┆ pipeline.execute()
except redis.exceptions.WatchError:
┆ pass
print conn.mget('a_num', "b_num")
結果：
['0', '20'] ['10', '10'] ['20', '0']
成功實現了賬戶轉移，但是有三次嘗試失敗了，如果要儘可能的使每次交易都獲得成功，可以加嘗試次數或者嘗試時間：
def a_to_b():
pipeline = conn.pipeline()
end = time.time() + 5
while time.time() < end:
┆ try:
┆ ┆ pipeline.watch('a_num')
┆ ┆ if int(pipeline.get('a_num')) < 10:
┆ ┆ ┆ pipeline.unwatch()
┆ ┆ ┆ return
┆ ┆ pipeline.multi()
┆ ┆ pipeline.decr('a_num', 10)
┆ ┆ pipeline.incr('b_num', 10)
┆ ┆ pipeline.execute()
┆ ┆ return True
┆ except redis.exceptions.WatchError:
┆ ┆ pass
return False
這樣，Redis可以使用事務實現類似於鎖的機制，但這個機制與關係型資料庫的鎖有所不同。關係型資料庫對被訪問的數據行進行加鎖時，其它客戶端嘗試對被加鎖數據行進行寫入是會被阻塞的。
Redis執行WATCH時並不會對數據進行加鎖，如果發現數據已經被其他客戶端搶先修改，只會通知執行WATCH命令的客戶端，並不會阻止修改，這稱之為樂觀鎖。
用SET()構建鎖
用WACTH實現的樂觀鎖一般情況下是適用的，但存在一個問題，程式會為完成一個執行失敗的事務而不斷地進行重試。當負載增加的時候，重試次數會上升到一個不可接受的地步。
如果要自己正確的實現鎖的話，要避免下麵幾個情況：
- 多個進程同時獲得了鎖
- 持有鎖的進程在釋放鎖之前崩潰了，而其他進程卻不知道
- 持有鎖的進行運行時間過長，鎖被自動釋放了，進程本身不知道，還會嘗試去釋放鎖
Redis中要實現鎖，需要用到一個命令，SET()或者說是SETNX()。SETNX只會在鍵不存在的情況下為鍵設置值，現在SET命令在加了NX選項的情況下也能實現這個功能，而且還能設置過期時間，簡直就是天生用來構建鎖的。
只要以需要加鎖的資源名為key設置一個值，要獲取鎖時，檢查這個key存不存在即可。若存在，則資源已被其它進程獲取，需要阻塞到其它進程釋放，若不存在，則建立key並獲取鎖：
import time
import uuid
class RedisLock(object):
def __init__(self, conn, lockname, retry_count=3, timeout=10,):
self.conn = conn
self.lockname = 'lock:' + lockname
self.retry_count = int(retry_count)
self.timeout = int(timeout)
self.unique_id = str(uuid.uuid4())
def acquire(self):
retry = 0
while retry < self.retry_count:
if self.conn.set(lockname, self.unique_id, nx=True, ex=self.timeout):
return self.unique_id
retry += 1
time.sleep(.001)
return False
def release(self):
if self.conn.get(self.lockname) == self.unique_id:
self.conn.delete(self.lockname)
return True
else:
return False
獲取鎖的預設嘗試次數限制3次，3次獲取失敗則返回。鎖的生存期限預設設為了10s，若不主動釋放鎖，10s後鎖會自動消除。
還保存了獲取鎖時鎖設置的值，當釋放鎖的時候，會先判斷保存的值和當前鎖的值是否一樣，如果不一樣，說明是鎖過期被自動釋放然後被其它進程獲取了。所以鎖的值必須保持唯一，以免釋放了其它程式獲取的鎖。
使用鎖：
def a_to_b():
lock = Redlock(conn, 'a_num')
if not lock.acquire():
┆ return False
pipeline = conn.pipeline()
try:
┆ pipeline.get('a_num')
┆ (a_num,) = pipeline.execute()
┆ if int(a_num) < 10:
┆ ┆ return False
┆ pipeline.decr('a_num', 10)
┆ pipeline.incr('b_num', 10)
┆ pipeline.execute()
┆ return True
finally:
┆ lock.release()
釋放鎖時也可以用Lua腳本來告訴Redis：刪除這個key當且僅當這個key存在而且值是我期望的那個值：
unlock_script = """
if redis.call("get",KEYS[1]) == ARGV[1] then
┆ return redis.call("del",KEYS[1])
else
┆ return 0
end"""
可以用conn.eval來運行Lua腳本：
def release(self):
┆ self.conn.eval(unlock_script, 1, self.lockname, self.unique_id)
這樣，一個Redis單機鎖就實現了。我們可以用這個鎖來代替WATCH，或者與WACTH同時使用。
實際使用中還要根據業務來決定鎖的粒度的問題，是鎖住整個結構還是鎖住結構中的一小部分。
粒度越大，性能越差，粒度越小，發生死鎖的幾率越大。
