如何实现Python异步函数短时间多次调用仅执行一次？

我正在开发一款Discord机器人，机器人内部包含一个server类（无复杂WebSocket与HTTP请求），以及作为用户与server之间桥梁的client类。client类实例负责向对应用户发送日志消息、更新GUI（仅包含一个embed及若干按钮），并调用server类的方法。

目前我在日志消息功能上遇到了问题：系统要求带控件的GUI消息必须始终是最新发送的消息。

当其他用户加入server类中的房间时，GUI消息无法再更新；此外，向用户发送日志消息也会导致GUI消息不再是最新消息。这两个问题都可以通过机器人删除旧GUI消息并发送更新后的消息解决。

但可能出现并发加入房间的情况，导致机器人更新GUI消息的「删除消息」与「发送消息」步骤交错执行，示例如下：

delete_message()
delete_message() # !!! 这里会因为找不到已删除的消息报错
send_message()
send_message()

我的拟解决方案对应以下技术问题：

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

import asyncio

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

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

通过foo_caller函数多次调用foo，当前会打印bar 7次。

现在需要实现：当foo在短时间内被多次调用时，仅执行一次函数调用。要求最终仅打印bar 3次：一次对应await asyncio.gather(foo(), foo(), foo(), foo(), foo())的批量调用，另外两次分别对应后续的两个await foo()调用。

可以通过维护「正在执行的任务」缓存结合异步锁，实现并发调用的合并。核心逻辑是：当foo被调用时，若已有未完成的执行任务，直接等待该任务结束即可，无需启动新任务。

代码实现如下：

import asyncio

# 缓存当前正在执行的foo任务
current_foo_task = None
# 异步锁，避免多任务并发修改任务缓存时出现竞态
foo_lock = asyncio.Lock()

async def foo():
    global current_foo_task
    async with foo_lock:
        # 检查是否有未完成的任务，有则直接等待
        if current_foo_task is not None and not current_foo_task.done():
            await current_foo_task
            return
        # 无正在执行的任务，创建并执行新任务
        async def _foo_impl():
            print("bar")
        current_foo_task = asyncio.create_task(_foo_impl())
        await current_foo_task
        # 任务完成后清空缓存
        current_foo_task = None

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

asyncio.run(foo_caller())

逻辑说明 #

1. 任务缓存：current_foo_task记录当前正在执行的foo任务实例，新调用进来时先检查该值。
2. 异步锁：foo_lock确保多任务并发访问任务缓存时的安全性，避免多个任务同时判断「无正在执行任务」并重复创建任务。
3. 并发合并：asyncio.gather中的5次foo()调用会依次进入锁，第一个调用创建任务并执行，后续4个调用发现已有任务，直接等待其完成，因此这5次调用仅触发一次print("bar")。
4. 单独调用处理：后续的两个await foo()调用时，前序任务已执行完成，缓存被清空，因此每个调用都会创建新任务并执行，各打印一次bar。

最终输出恰好为bar 3次，满足需求。

内容的提问来源于stack exchange，提问作者TNTzx