如何在自定义子类中正确执行外部类的异步函数？

问题根源 #

你遇到的RuntimeWarning是因为调用异步函数funcB时未使用await关键字，也未通过异步事件循环执行这个协程对象。异步函数调用后会返回协程对象，必须被等待或放入事件循环才能实际执行。

另外你的代码存在几处语法错误，先修正基础问题：

• 类C的初始化方法应为__init__（双下划线），而非init
• funcE缩进错误，需与__init__同级
• 变量名variablel拼写错误，修正为variable1

解决方案 #

方案1：将funcE改为异步方法 #

若代码场景允许异步调用，直接把funcE改为异步方法，内部用await调用funcB，再通过事件循环启动：

import asyncio

# 外部包类，不可修改
class A:
    def funcA(self):
        print("Something from funcA")

    async def funcB(self):
        print("Something from funcB")

# 外部包类，不可修改
class B:
    def funcC(self):
        print("Something from funcC")

    def funcD(self):
        self.funcC()
        print("Something from funcD")

# 自定义类
class C(B):
    def __init__(self):
        self.variable1 = A()

    async def funcE(self):
        await self.variable1.funcB()

# 启动异步执行
asyncio.run(C().funcE())

方案2：在同步方法内用asyncio.run执行异步函数 #

若必须保持funcE为同步方法（比如兼容B类的同步调用逻辑），可在funcE内部用asyncio.run()运行异步协程：

import asyncio

# 外部包类，不可修改
class A:
    def funcA(self):
        print("Something from funcA")

    async def funcB(self):
        print("Something from funcB")

# 外部包类，不可修改
class B:
    def funcC(self):
        print("Something from funcC")

    def funcD(self):
        self.funcC()
        print("Something from funcD")

# 自定义类
class C(B):
    def __init__(self):
        self.variable1 = A()

    def funcE(self):
        asyncio.run(self.variable1.funcB())

D = C()
D.funcE()

方案3：线程运行事件循环（适配复杂同步场景） #

若同步代码不能阻塞（如GUI线程、高并发场景），可开启线程运行事件循环，再提交协程执行：

import asyncio
import threading

# 外部包类，不可修改
class A:
    def funcA(self):
        print("Something from funcA")

    async def funcB(self):
        print("Something from funcB")

# 外部包类，不可修改
class B:
    def funcC(self):
        print("Something from funcC")

    def funcD(self):
        self.funcC()
        print("Something from funcD")

# 自定义类
class C(B):
    def __init__(self):
        self.variable1 = A()
        # 启动异步事件循环线程
        self.loop = asyncio.new_event_loop()
        self.thread = threading.Thread(target=self.loop.run_forever, daemon=True)
        self.thread.start()

    def funcE(self):
        # 提交协程到事件循环执行
        asyncio.run_coroutine_threadsafe(self.variable1.funcB(), self.loop)

D = C()
D.funcE()

注意事项 #

• 纯异步项目优先选方案1，保持代码风格统一
• 方案2简单直接，但asyncio.run()每次调用都会创建/关闭事件循环，频繁调用有性能损耗
• 方案3适合同步环境下持续执行异步任务的场景，需注意线程安全问题

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