Python如何实现非阻塞的独立异步函数循环执行？

你的核心问题是不想让FunctionB的处理阻塞FunctionA的持续执行，同时还要避免无限制创建任务导致的资源泄漏。可以通过维护任务列表、定期清理已完成任务的方式解决，具体实现如下：

核心修改思路 #

1. 用列表追踪所有正在运行的FunctionB任务
2. 创建任务后直接放入列表，不立即await，让任务在后台异步执行
3. 每次循环时清理已完成的任务，避免内存占用过高
4. 给事件循环留调度时间，防止FunctionA无限制循环占用CPU

修改后的代码示例 #

import asyncio

def functionA():
    # 模拟同步获取信息的逻辑（如果是IO密集型建议改成异步）
    current_time = asyncio.get_event_loop().time()
    print(f"获取到数据: {round(current_time, 2)}")
    return current_time  # 模拟有效数据返回

async def functionB(data):
    # 模拟耗时的异步处理逻辑
    print(f"开始处理数据: {round(data, 2)}")
    await asyncio.sleep(2)  # 替换为实际的耗时操作
    print(f"完成处理数据: {round(data, 2)}")

async def main():
    running_tasks = []
    while True:
        data = functionA()
        if data:
            # 创建异步任务并加入列表
            task = asyncio.create_task(functionB(data))
            running_tasks.append(task)
        
        # 清理已完成的任务，释放资源
        running_tasks = [task for task in running_tasks if not task.done()]
        
        # 给事件循环时间调度其他任务，同时避免FunctionA循环过快
        await asyncio.sleep(0.1)

asyncio.run(main())

额外优化建议 #

如果FunctionA本身是IO密集型操作（比如网络请求、文件读取），建议将其也改为异步函数，进一步提升整体效率：

async def functionA():
    # 模拟异步获取信息（比如异步HTTP请求）
    await asyncio.sleep(0.5)
    current_time = asyncio.get_event_loop().time()
    print(f"异步获取到数据: {round(current_time, 2)}")
    return current_time

async def main():
    running_tasks = []
    while True:
        data = await functionA()
        if data:
            task = asyncio.create_task(functionB(data))
            running_tasks.append(task)
        
        running_tasks = [task for task in running_tasks if not task.done()]

为什么这样可行？ #

• asyncio.create_task会将FunctionB任务提交给事件循环，让它在后台独立执行，不会阻塞main循环的后续迭代
• 任务列表的清理逻辑确保不会积累大量已完成的任务，避免内存泄漏
• await asyncio.sleep(0.1)（或await functionA()）会触发事件循环切换到其他待执行的任务，保证FunctionB能被正常调度

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