OOP设计中使用asyncio.wait遇任务执行顺序问题求助

问题出在两个核心点：

• asyncio.run参数错误：你传入的是协程函数self.aws_wrapper，但asyncio.run需要的是调用后的协程对象，必须加括号写成self.aws_wrapper()——这就是触发"An asyncio.Future, a coroutine or an awaitable is required"错误的直接原因。
• 执行顺序未同步：就算参数改对，也要确保异步拉取任务完全完成后再执行业务逻辑，当前写法的流程没有卡死这个顺序。

修正后的完整代码 #

import asyncio

# 模拟AWS异步客户端（实际项目替换为真实异步AWS SDK，如aiobotocore）
class MockAWS:
    async def get_stuff(self, stuff):
        # 模拟网络IO延迟
        await asyncio.sleep(1)
        return f"{stuff}的拉取数据"

aws = MockAWS()

class Application:
    def __init__(self, A, B):
        self.A = A
        self.B = B
        self.a_result = None  # 存储A的拉取结果
        self.b_result = None  # 存储B的拉取结果

    async def pull(self, stuff):
        # 直接await异步AWS调用，无需多余赋值步骤
        return await aws.get_stuff(stuff)

    def applicationLogic(self):
        # 业务逻辑可直接使用拉取到的结果
        print("开始执行业务逻辑")
        print(f"A的拉取结果: {self.a_result}")
        print(f"B的拉取结果: {self.b_result}")

    async def aws_wrapper(self):
        # 显式创建异步任务
        a_task = asyncio.create_task(self.pull(self.A))
        b_task = asyncio.create_task(self.pull(self.B))
        # 等待所有任务完成
        done, pending = await asyncio.wait([a_task, b_task])
        # 提取任务结果并保存到实例属性
        for task in done:
            if task is a_task:
                self.a_result = task.result()
            elif task is b_task:
                self.b_result = task.result()

    def aws_async_executor(self):
        # 修正：传入协程对象（调用aws_wrapper）
        asyncio.run(self.aws_wrapper())
        # 此时异步任务已全部完成，再执行业务逻辑
        self.applicationLogic()

# 测试运行
app = Application("资源A", "资源B")
app.aws_async_executor()

关键改进说明 #

1. 参数修正：把asyncio.run(self.aws_wrapper)改为asyncio.run(self.aws_wrapper())，确保asyncio能正确识别并执行协程。
2. 结果留存：用asyncio.create_task创建任务，等待完成后将结果存储到实例变量，让业务逻辑可以直接复用拉取到的数据。
3. 顺序保证：asyncio.run会阻塞直到传入的协程（aws_wrapper）完全执行完毕，之后调用applicationLogic时，异步拉取任务肯定已经完成。
4. AWS调用优化：如果使用异步AWS SDK，pull方法直接await接口调用即可，避免多余的赋值操作。

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