這是我2019年開發的，已經在很多項目中實踐認證，很好用
import time
import redis
"""
基於Redis實現的分布式悲觀鎖
@site http://itsub.cn
@author 夏增明
"""
class RedisLock():
def __init__(self, ):
self.conn = redis.Redis(host='', port=6379, password="", db=1)
def acquire_lock(self, lockname, identifier='ok', expire=10):
conn = self.conn
if conn.setnx(lockname, identifier):
conn.expire(lockname, expire)
return identifier
elif not conn.ttl(lockname):
conn.expire(lockname, expire)
return False
def release_lock(self, lockname, identifier='ok'):
conn = self.conn
pipe = conn.pipeline(True)
while True:
try:
pipe.watch(lockname)
if pipe.get(lockname) == identifier:
pipe.multi()
pipe.delete(lockname)
pipe.execute()
return True
pipe.unwatch()
break
except redis.exceptions.WatchError:
pass
# we lost the lock
return False
def lock(lockname,expire=10):
lock = RedisLock()
def _deco(func):
def __deco(*args, **kwargs):
#print( "before %s called [%s]." % (func.__name__, lock) )
lock.acquire_lock(lockname,expire=expire)
try:
return func(*args, **kwargs)
finally:
lock.release_lock(lockname)
return __deco
return _deco
#通過鎖名字進行排隊同步
@lock('test')
def hello():
print("hello() called.")
time.sleep(20)
if __name__ == "__main__":
hello()好東西不私藏，點贊關注呗
