我在用aiogram机器人中的中间件为用户实现一个速率限制器。现在我为请求定义了一个类，并为速率限制器定义了一个类。我的问题是，在性能和安全性之间，应该对每个操作都使用多次异步锁，还是把这些操作放在同一个异步锁中执行。

我的rate_limiter.py文件：

import asyncio
import time
from collections import defaultdict
from dataclasses import dataclass

class RequestsData:
    def __init__(
        self,
        seconds_interval: int = 3,
        max_requests_per_day: int = 5,
    ):
        self.seconds_interval: int = seconds_interval
        self.max_requests_per_day: int = max_requests_per_day
        self.day_len_seconds: int = 60 * 60 * 24

        self.last_time: float | None = None
        self.just_limit_exited: bool = True

        self.total_requests: int = 0
        self.total_requests_start_time: float | None = None

        self._async_lock = asyncio.Lock()

    @classmethod
    def get_now_seconds(self) -> float:
        return time.time()

    async def reset_day_limit(self) -> bool:
        async with self._async_lock:
            # no requests start time
            if self.total_requests_start_time is None:
                return True

            current_time_seconds = self.get_now_seconds()
            # check and reset total requests start time as current time seconds
            if current_time_seconds - self.total_requests_start_time > self.day_len_seconds:
                self.total_requests_start_time = current_time_seconds
                self.total_requests = 0
                return True

            return False

    async def exited_day_limit(self) -> tuple[bool, bool]:
        async with self._async_lock:
            if self.total_requests >= self.max_requests_per_day-1:
                # set and reuturn limit exited
                if self.just_limit_exited:
                    self.just_limit_exited = False

                    # limit exited, not blocked
                    return (True, False)
                return (False, False)
            # return no exited, can make requests
            return (False, True)

    async def rate_limited(self) -> tuple[bool, bool]:
        async with self._async_lock:
            current_time_seconds = self.get_now_seconds()

            # check times diff less than interval
            if self.last_time is not None and current_time_seconds - self.last_time < self.seconds_interval:
                # set and return limit exited
                if self.just_limit_exited:
                    self.just_limit_exited = False

                    # limit exited and blocked
                    return (True, False)

                # limit not exited and blocked
                return (False, False)

            # limit not exited and not blocked
            return (False, True)

    async def add_reuqest(self):
        async with self._async_lock:
            current_time_seconds = self.get_now_seconds()

            self.last_time = current_time_seconds
            self.just_limit_exited = True
            self.total_requests += 1


class RateLimiter:
    def __init__(self, 
        seconds_interval: int=1,
        max_requests_per_day: int=5,
    ):
        self.requsts = defaultdict[int, RequestsData](
            lambda: RequestsData(seconds_interval, max_requests_per_day)
        )

    async def on_request(self, uid: int) -> tuple[bool, bool]:
        # get requests data by uid
        uid_requests: RequestsData = self.requsts[uid]

        # check and reset day limit start
        await uid_requests.reset_day_limit()

        # check exited day limit    
        just_exited, can_make_requests = await uid_requests.exited_day_limit()

        if not can_make_requests:
            return (just_exited, can_make_requests)

        # check rate limited by last time
        just_exited, can_make_requests = await uid_requests.rate_limited()

        if not can_make_requests:
            return (just_exited, can_make_requests)

        # add reuqest with last time as now
        await uid_requests.add_reuqest()

        # can make requests, not blocked
        return (False, True)

我在中间件中使用这个速率限制器：

class MainMiddleware(BaseMiddleware):
    def __init__(self):
        super().__init__()

        # set rate limiter
        self.rate_limiter = RateLimiter()

    async def __call__(
            self, 
            handler: Callable[[TelegramObject, Dict[str, Any]],  Awaitable[Any]], 
            event: Message | CallbackQuery, 
            data: Dict[str, Any]
        ):

        # check chat type is private
        if isinstance(event, Message):
            chat_type = event.chat.type
        else:
            chat_type = event.message.chat.type

        if chat_type != ChatType.PRIVATE:
            logger.info("chat is not private %s", event)
            return 

        user_id = event.from_user.id
        lang = event.from_user.language_code

        # check with rete limiter
        just_limit_exited, can_make_request = await self.rate_limiter.on_request(
            user_id
        )

        # notify can't make requests
        if just_limit_exited:
            text = get_text(texts.rate_limit_exited_text, lang)
            await event.answer(text)

        # cant make requests
        if not can_make_request:
            return None

解决方案

你可以删除初始代码块中的所有 async 和 await，包括锁。

asyncio 使用协作式多任务模型。所有 asyncio 任务都在同一线程上运行，且一个 async 函数 不会被中断，除非你明确 await 某事（并且只有在你明确 await 一个任务时，才会启动其他任务）。

这意味着，如果你有一个只是进行纯计算的函数，它在 asyncio 的情况下也不会被中断。这描述了 RequestData 类中所有函数的行为。所以当你获得 asyncio 锁时，函数在退出锁之前不会被中断，而锁基本上也没什么作用——没有任何时刻会有另一个 async 函数开始运行并可能调用同一个对象。如果你移除了对锁的调用，那么在函数中就没有 async 或 await，并且在被调用时函数也不需要 async、或 awaited。

类似地，RateLimiter.on_request() 如果你做出这个改变就不再需要 async（它不再调用任何 async 函数）。中间件函数可能需要是 async，并且可能会被多次调用（每个并发请求一次），但按一般的 asyncio 模型，同一时刻只会有一个中间件函数及其底层的速率限制器在运行。

这与多线程不同，在多线程的场景中，理论上可能有两个不同的线程同时访问同一个对象，像你展示的那种锁定模式可能是必要的。asyncio 本质上是单线程的，这会导致需要的锁更少。