快捷導(dǎo)航

Python+Redis從零打造分布式鎖實(shí)戰(zhàn)示例

更新時間：2024年01月22日 08:51:07 作者：傻子的尷尬 IT智慧谷

Redis作為一款高性能的內(nèi)存鍵值數(shù)據(jù)庫,憑借其支持原子操作、高并發(fā)和數(shù)據(jù)持久化等特性,非常適合用來實(shí)現(xiàn)分布式鎖,本文將詳細(xì)探討如何使用Python結(jié)合Redis從簡單到復(fù)雜地實(shí)現(xiàn)分布式鎖,并提供相應(yīng)的示例代碼

引言

在分布式系統(tǒng)中，多個節(jié)點(diǎn)可能需要訪問同一共享資源，這就需要一種協(xié)調(diào)機(jī)制來保證在同一時刻只有一個節(jié)點(diǎn)進(jìn)行操作，這就是分布式鎖的作用。

1. 簡單實(shí)現(xiàn)（基于SETNX命令）

使用setnx命令（set if not exists）嘗試設(shè)置一個key-value對，如果key不存在，則設(shè)置成功并返回1，表示獲取到了鎖。

import redis
import time
def acquire_lock(redis_client, lock_key):
    identifier = str(time.time()) + str(id(threading.current_thread()))
    acquired = redis_client.setnx(lock_key, identifier)
    return acquired
def release_lock(redis_client, lock_key, identifier):
    pipe = redis_client.pipeline(True)
    while True:
        try:
            pipe.watch(lock_key)
            if pipe.get(lock_key) == identifier:
                pipe.multi()
                pipe.delete(lock_key)
                pipe.execute()
                return True
            pipe.unwatch()
            break
        except redis.exceptions.WatchError:
            pass
    return False
# 使用示例
r = redis.Redis(host='localhost', port=6379, db=0)
lock_key = 'mylock'
if acquire_lock(r, lock_key):
    print("Lock acquired")
    # 執(zhí)行臨界區(qū)代碼...
    release_lock(r, lock_key, identifier)

2. 改進(jìn)：設(shè)置超時時間（expire命令）

為防止進(jìn)程崩潰導(dǎo)致鎖無法釋放，可以給鎖設(shè)置一個過期時間。

def acquire_lock_with_timeout(redis_client, lock_key, expire_time):
    identifier = str(time.time()) + str(id(threading.current_thread()))
    while True:
        result = redis_client.set(lock_key, identifier, nx=True, ex=expire_time)
        if result:
            return identifier
        time.sleep(0.1)
# 使用示例不變，但在acquire_lock_with_timeout函數(shù)中增加了expire_time參數(shù)

3. 進(jìn)一步改進(jìn)：使用Lua腳本保證原子性

在釋放鎖的時候，需要確保刪除的是自己設(shè)置的鎖，且這個操作是原子的?？梢允褂肔ua腳本來實(shí)現(xiàn)。

RELEASE_LOCK_SCRIPT = """
if redis.call("get", KEYS[1]) == ARGV[1] then
    return redis.call("del", KEYS[1])
else
    return 0
end
"""
def release_lock LUA(redis_client, lock_key, identifier):
    release_script = redis_client.register_script(RELEASE_LOCK_SCRIPT)
    released = release_script(keys=[lock_key], args=[identifier])
    return bool(released)
# 使用示例不變

4. 更完善的實(shí)現(xiàn)，Redlock算法

Redlock算法的實(shí)現(xiàn)相對復(fù)雜，以下是一個簡化的Python示例，展示了如何在多個Redis實(shí)例上嘗試獲取和釋放鎖。請注意，在實(shí)際應(yīng)用中，需要確保Redis實(shí)例之間的時間同步，并且考慮網(wǎng)絡(luò)延遲等因素。

import redis
import time
class RedLock:
    def __init__(self, masters, ttl):
        self.masters = [redis.Redis(host=host, port=port, db=db) for host, port, db in masters]
        self.quorum = len(self.masters) // 2 + 1
        self.ttl = ttl
    def lock(self, resource_id):
        expiration = time.time() + self.ttl
        while time.time() < expiration:
            acquired_count = 0
            locked_instances = []
            # 嘗試在每個Redis實(shí)例上獲取鎖
            for master in self.masters:
                if master.set(resource_id, 'locked', nx=True, px=self.ttl):
                    acquired_count += 1
                    locked_instances.append(master)
            # 如果獲得超過半數(shù)以上的鎖，則認(rèn)為成功獲取分布式鎖
            if acquired_count > self.quorum:
                return locked_instances
            # 清除已獲取但未達(dá)到法定數(shù)量的鎖
            for master in locked_instances:
                master.delete(resource_id)
            # 等待一段時間后重試
            time.sleep(0.1)
        raise Exception("Could not acquire lock")
    def unlock(self, resource_id, masters_with_lock):
        for master in masters_with_lock:
            # 使用Lua腳本保證原子性地刪除鎖（與前面簡單實(shí)現(xiàn)中的釋放鎖方法類似）
            RELEASE_LOCK_SCRIPT = """
            if redis.call("get", KEYS[1]) == "locked" then
                return redis.call("del", KEYS[1])
            else
                return 0
            end
            """
            release_script = master.register_script(RELEASE_LOCK_SCRIPT)
            release_script(keys=[resource_id])
# 示例用法：
masters = [('localhost', 6379, 0), ('localhost', 6380, 0)]  # 假設(shè)有兩個Redis實(shí)例
redlock = RedLock(masters, 10000)  # 設(shè)置鎖的有效期為10秒
resource_id = 'my_resource'
try:
    masters_with_lock = redlock.lock(resource_id)
    print(f"Acquired Redlock on resource {resource_id}")
    # 執(zhí)行臨界區(qū)代碼...
finally:
    redlock.unlock(resource_id, masters_with_lock)

此示例僅為了說明Redlock算法的核心思想，并未涵蓋所有邊界條件和異常處理，如Redis實(shí)例宕機(jī)、網(wǎng)絡(luò)延遲導(dǎo)致時間不一致等情況，請在生產(chǎn)環(huán)境中使用時結(jié)合實(shí)際情況進(jìn)行優(yōu)化和完善。

總結(jié)

通過以上介紹和示例代碼，我們了解了如何基于Python與Redis實(shí)現(xiàn)不同層次復(fù)雜度的分布式鎖。從簡單的SETNX命令獲取鎖，到設(shè)置鎖的超時時間以防止進(jìn)程崩潰導(dǎo)致死鎖，再到利用Lua腳本保證釋放鎖操作的原子性，以及針對大規(guī)模分布式環(huán)境考慮采用Redlock算法提高系統(tǒng)的容錯性和安全性。這些方法可以根據(jù)實(shí)際業(yè)務(wù)場景選擇合適的方式來實(shí)現(xiàn)分布式鎖，從而有效解決多節(jié)點(diǎn)環(huán)境下對共享資源的競爭問題。隨著技術(shù)的發(fā)展和需求的變化，分布式鎖的實(shí)現(xiàn)方式也將持續(xù)演進(jìn)和完善，為構(gòu)建更穩(wěn)定、高效的分布式系統(tǒng)提供有力支撐。

以上就是Python+Redis從零打造分布式鎖實(shí)戰(zhàn)示例的詳細(xì)內(nèi)容，更多關(guān)于Python Redis分布式鎖的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: