Python：当函数在短时间内被多次调用时，如何只执行一个异步函数调用？

Question

语境：

我目前正在为 Discord 上的机器人编写代码。该机器人内部有一个服务器类（没有花哨的 websockets 和 http 请求）和一个充当用户和服务器之间桥梁的客户端类。客户端类的实例管理向其相应用户发送日志消息、更新其 GUI（这只是一个嵌入和附加到它的一堆按钮）以及调用服务器类上的方法。

目前我被困在日志消息上。当前系统是包含控件的 GUI 消息始终是最近发送的消息。

如果另一个用户要加入服务器类的房间，这将导致 GUI 消息不再更新。此外，将向用户发送一条日志消息，这将导致 GUI 消息不是最近发送的消息。这两个问题都可以通过机器人删除旧的 GUI 消息并在此之后发送更新的消息来解决。

但是，可能会同时加入房间，因此机器人有可能会像这样将更新 GUI 消息的“删除消息”和“发送消息”部分交错：

delete_message()
delete_message() # !!!
send_message()
send_message()

第二个delete_message() 会导致错误，因为它找不到已删除的消息。

我提出的解决方案将是以下问题。

---

问题：

假设我有一个名为 foo 的异步函数：

import asyncio


limit: int

async def foo():
    print("bar")


async def foo_caller():
    await asyncio.gather(foo(), foo(), foo(), foo(), foo())
    await foo()
    await foo()

该函数将使用foo_caller 函数多次调用。目前，这将打印bar7次.

问题是，如何在短时间内多次调用foo 时只执行一个函数调用？

解决方案应仅打印bar三次.一个用于await asyncio.gather(foo(), foo(), foo(), foo(), foo())，一个用于await foo()。

Answer 1

在下面回答您的问题，但我认为您在这里遇到 X-> Y 问题，您只需要使用消息 ID。但我也会回答限制异步执行。

你可以使用一个基于时间的锁，比如说 5 秒。我们可以确定所有的请求都将在 5 秒内发送出去。

async def foo():
    if lock_exists():
        return
    async with lock(ttl=5):  # lock is alive for 5 seconds, the rest don't execute. Not released upon execution, but on a timer
        print("bar")


async def foo_caller():
    await asyncio.gather(foo(), foo(), foo(), foo(), foo())
    await foo()
    await foo()

但它是客户端逻辑，您可以在尝试删除或 Try-Except 之前检查它是否存在。

另一种方法是可以例外 -

async def foo():
    print("bar")


async def foo_caller():
    await asyncio.gather(foo(), foo(), foo(), foo(), foo(), return_exceptions=True) # Exceptions don't stop the code
    await foo()
    await foo()

这不会引发异常。 但是我认为您实际上需要它来记录您需要删除的消息的ID。所以

def delete_message(id_):  # get the id with a get request or similar
    send_delete_request_to_discord_here(message_id=id_)

也许在每个 send_message 之前获取要删除的消息的存储空间或执行获取请求以获取通道中的最后一条消息。如果存在，则仅将其删除。类似的东西。

Answer 2

这是一个类“Regulator”，可用于以符合您要求的方式包装任何 Callable。在给定的时间间隔内，该函数永远不会被多次调用。多余的调用将被丢弃。

main 功能与您的foo_caller 几乎相同，但我添加了一些时间延迟，以便清楚测试程序有效。程序打印了三次“bar”。

import asyncio
from typing import Callable

class Regulator:
    def __init__(self, interval: float, f: Callable, *args, **kwargs):
        """
        Do not call the function f more than one per t seconds.
        
        interval is a time interval in seconds.
        f is a function
        *args and **kwargs are the usual suspects
        """
        self.interval = interval
        self.f = f
        self.args = args
        self.kwargs = kwargs
        self.busy = False
        
    async def __call__(self):
        if not self.busy:
            self.busy = True
            asyncio.get_event_loop().call_later(self.interval, self.done)
            self.f(*self.args, **self.kwargs)
            
    def done(self):
        self.busy = False
        
def say_bar():
    print("bar")
            
foo = Regulator(0.5, say_bar)

async def main():
    await asyncio.gather(foo(), foo(), foo(), foo(), foo())
    await asyncio.sleep(1.0)
    await foo()
    await asyncio.sleep(1.0)
    await foo()

if __name__ == "__main__":
    asyncio.run(main())