class Counter(object):
    def __init__(self, namespace: str) -> None:
        self.namespace: int = namespace
        self.count: int = 0
    def login(self) -> None:
        self.count += 1
    def logout(self) -> None:
        self.count -= 1

from threading import Lock
class Counter(object):
    def __init__(self, namespace: str) -> None:
        self.namespace: int = namespace
        self.count: int = 0
        self._lock: Lock = Lock()
    def login(self) -> None:
        with lock():
            self.count += 1
    def logout(self) -> None:
        with lock():
            self.count -= 1

import asyncio
from redis.asyncio import Redis
class Counter(object):
    def __init__(self, namespace: str, redis: Redis) -> None:
        self._redis = redis
        self.namespace: int = namespace
        self.count: int = 0
    async def login(self) -> None:
        async with self._redis.lock("demo"):
            self.count += 1
    async def logout(self) -> None:
        async with self._redis.lock("demo"):
            self.count -= 1
async def main():
    counter = Counter("demo", Redis())
    await counter.login()
    await counter.logout()
asyncio.run(main())

class Lock:
    async def __aenter__(self):
        pass
    async def __aexit__(self, exc_type, exc_value, traceback):
        pass
    async def acquire(
        self,
        blocking: Optional[bool] = None,
        blocking_timeout: Optional[float] = None,
        token: Optional[Union[str, bytes]] = None,
    ):
        pass
    async def locked(self) -> bool:
        pass
    async def owned(self) -> bool:
        pass
    def release(self) -> Awaitable[None]:
        pass

class Lock:
    ...
    async def __aenter__(self):
        if await self.acquire():
            return self
        raise LockError("Unable to acquire lock within the time specified")
    async def __aexit__(self, exc_type, exc_value, traceback):
        await self.release()

class Lock:
    ...
    async def acquire(
        self,
        blocking: Optional[bool] = None,
        blocking_timeout: Optional[float] = None,
        token: Optional[Union[str, bytes]] = None,
    ):
        ################################
        # 第一部分，這里主要是初始化各種參數(shù) #
        ################################
        sleep = self.sleep
        if token is None:
            token = uuid.uuid1().hex.encode()
        else:
            try:
                encoder = self.redis.connection_pool.get_encoder()
            except AttributeError:
                # Cluster
                encoder = self.redis.get_encoder()
            token = encoder.encode(token)
        if blocking is None:
            blocking = self.blocking
        if blocking_timeout is None:
            blocking_timeout = self.blocking_timeout
        stop_trying_at = None
        if blocking_timeout is not None:
            stop_trying_at = asyncio.get_running_loop().time() + blocking_timeout
        #####################################################
        # 第二部分，這里通過一個循環(huán)去爭奪鎖，在爭奪成功時會返回True #
        #####################################################
        while True:
            if await self.do_acquire(token):
                # 爭奪成功，把Token存入當前線程存儲中，并返回True
                self.local.token = token
                return True
            # 爭奪失敗且設(shè)置blocking為False，則返回獲取鎖失敗
            if not blocking:
                return False
            # 在設(shè)置的時間內(nèi)爭奪失敗，返回獲取鎖失敗
            next_try_at = asyncio.get_running_loop().time() + sleep
            if stop_trying_at is not None and next_try_at > stop_trying_at:
                return False
            # 每次爭奪鎖的間隔，這是在初始化Lock對象時指定的，
            # 建議根據(jù)業(yè)務(wù)設(shè)置比較大的sleep時間，防止獲取鎖失敗時頻繁的與Redis通信。
            await asyncio.sleep(sleep)
    async def do_acquire(self, token: Union[str, bytes]) -> bool:
        if self.timeout:
            # convert to milliseconds
            timeout = int(self.timeout * 1000)
        else:
            timeout = None
        # 真正通過Redis設(shè)置鎖的方法，
        if await self.redis.set(self.name, token, nx=True, px=timeout):
            return True
        return False

class Lock:
    ...
    # 初始化時會初始化lua腳本--`LUA_RELEASE_SCRIPT`，
    # 并把Redis返回的ID存放到lua_release中，減少后續(xù)調(diào)用的網(wǎng)絡(luò)傳輸
    lua_release = None
    # lua腳本的邏輯是通過name獲取token，如果token不存在或者不是實例產(chǎn)生的則返回0，
    # 否則代表成功獲取到鎖，然后會執(zhí)行刪除鎖并返回1告訴客戶端說鎖釋放成功。
    LUA_RELEASE_SCRIPT = """
        local token = redis.call('get', KEYS[1])
        if not token or token ~= ARGV[1] then
            return 0
        end
        redis.call('del', KEYS[1])
        return 1
    """
    ...
    def release(self) -> Awaitable[None]:
        # 如果沒有從本地線程存儲獲取到Token，則代表還沒有獲取到鎖，拋出對應(yīng)的異常
        expected_token = self.local.token
        if expected_token is None:
            raise LockError("Cannot release an unlocked lock")
        # 先清楚本地線程存放的Token，再釋放鎖
        self.local.token = None
        return self.do_release(expected_token)
    async def do_release(self, expected_token: bytes) -> None:
        # 調(diào)用lua腳本并根據(jù)響應(yīng)結(jié)果判斷釋放鎖是否成功。
        if not bool(
            await self.lua_release(
                keys=[self.name], args=[expected_token], client=self.redis
            )
        ):
            raise LockNotOwnedError("Cannot release a lock that's no longer owned")

class Lock:
    # 效果與lua_release一致 
    lua_extend = None
    # 續(xù)約的Lua代碼主要邏輯為：
    # 1.判斷token是否一致，不一致則不是自己獲取的鎖，返回0。
    # 2.判讀鎖的過期時間，如果小于0則鎖不存在，返回0
    # 3.如果當前鎖的過期時間還未變?yōu)?,則為當前的鎖續(xù)約
    #   續(xù)約的方式有兩種，一種是在原來的基礎(chǔ)增加一個固定值，另一種是把過期時間設(shè)置為固定值。
    #   最后再返回1，代表續(xù)約成功
    LUA_EXTEND_SCRIPT = """
        local token = redis.call('get', KEYS[1])
        if not token or token ~= ARGV[1] then
            return 0
        end
        local expiration = redis.call('pttl', KEYS[1])
        if not expiration then
            expiration = 0
        end
        if expiration < 0 then
            return 0
        end
        local newttl = ARGV[2]
        if ARGV[3] == "0" then
            newttl = ARGV[2] + expiration
        end
        redis.call('pexpire', KEYS[1], newttl)
        return 1
    """
    def extend(
        self, additional_time: float, replace_ttl: bool = False
    ) -> Awaitable[bool]:
        # 判斷是否持有鎖，沒有鎖不能續(xù)約
        if self.local.token is None:
            raise LockError("Cannot extend an unlocked lock")
        # 沒設(shè)置超時不能續(xù)約
        if self.timeout is None:
            raise LockError("Cannot extend a lock with no timeout")
        return self.do_extend(additional_time, replace_ttl)
    async def do_extend(self, additional_time, replace_ttl) -> bool:
        # 通過Lua腳本執(zhí)行續(xù)約的邏輯
        additional_time = int(additional_time * 1000)
        if not bool(
            await self.lua_extend(
                keys=[self.name],
                args=[self.local.token, additional_time, replace_ttl and "1" or "0"],
                client=self.redis,
            )
        ):
            raise LockNotOwnedError("Cannot extend a lock that's no longer owned")
        return True

import asyncio
from typing import Optional, Union
from redis.asyncio.lock import Lock
class MyLock(Lock):
    # 存儲循環(huán)的協(xié)程對象
    _watch_dog: Optional[asyncio.Task]
    async def _watch(self) -> None:
        """
        這是一個一直在循環(huán)的方法，它每次循環(huán)都會執(zhí)行續(xù)約，然后休眠超時時間的1/3，然后再執(zhí)行下一個循環(huán)。
        """
        while True:
            await self.extend(self.timeout)
            await asyncio.sleep(self.timeout / 3)
    def _cancel_watch_dog(self) -> None:
        """
        取消正在運行WatchDog的協(xié)程
        """
        _old_watch_dog: Optional[asyncio.Future] = getattr(self, "_watch_dog", None)
        if _old_watch_dog and not _old_watch_dog.cancelled():
            _old_watch_dog.cancel()
    async def acquire(
        self,
        blocking: Optional[bool] = None,
        blocking_timeout: Optional[float] = None,
        token: Optional[Union[str, bytes]] = None,
    ) -> bool:
        result = await super().acquire(blocking, blocking_timeout, token)
        if result:
            # 獲取鎖成功就開始啟用WatchDog機制
            # 它先把舊的WatchDog取消，再通過`asyncio.create_task`方法使用一個單獨的協(xié)程執(zhí)行WatchDog
            self._cancel_watch_dog()
            self._watch_dog = asyncio.create_task(self._watch())
        return result
    async def do_release(self, expected_token: bytes) -> None:
        # 在釋放之前取消WatchDod機制
        # 即使取消TachDog成功，但是鎖釋放失敗也沒關(guān)系，因為有超時機制兜底
        self._cancel_watch_dog()
        return await super().do_release(expected_token)
    def __del__(self) -> None:
        try:
            self._cancel_watch_dog()
        except Exception:
            pass

import asyncio
import time
from typing import Optional, Union
from redis.asyncio.lock import Lock
from redis.asyncio import Redis
# WatchDog實現(xiàn)省略...
async def main():
    _redis = Redis()
    s_t = time.time()
    # pyredis的`lock`方法通過`lock_class`參數(shù)支持自定義Lock類
    async with _redis.lock("demo", lock_class=MyLock, timeout=3):
        print("lock")
        await asyncio.sleep(5)
    print("ok", time.time() - s_t)
if __name__ == "__main__":
    asyncio.run(main())