如何在Python不同类中同时运行asyncio异步任务

原代码存在的明确错误 #

• 类内错误放置入口判断：if __name__ == "__main__"块写在FirstClass类定义内部，既不属于类方法也不属于类属性，执行逻辑完全不符合预期
• 线程target传参错误：target=s_class.display_info(self)这类写法是直接在主线程同步调用对应方法，会直接卡死在无限循环中，根本不会启动新线程执行任务
• 协程无法直接作为线程任务运行：f_class.main()返回的是协程对象，普通线程默认没有启动asyncio事件循环，直接传入不会执行任何异步逻辑
• 全局事件循环耦合过高：跨类、跨线程引用全局loop极易出现事件循环绑定、上下文不一致的报错，且await asyncio.sleep(20, loop=loop)的写法在Python 3.8+版本已被废弃，不需要手动传入loop参数
• 阻塞死循环无休眠：display_info的无限循环没有任何等待逻辑，启动后会直接占满单个CPU核心资源

推荐实现（asyncio统一调度，无需手动管理线程） #

优先用asyncio原生能力调度任务，把同步阻塞方法扔到自带线程池执行，不需要手动维护线程生命周期，稳定性更高、代码更简洁：

import asyncio
import time

class FirstClass:
    async def main(self):
        print("Hello 异步任务启动")
        while True:
            print("异步任务运行中，距上次输出间隔20s")
            await asyncio.sleep(20)

class SecondClass:
    def display_info(self):
        while True:
            print('People 同步阻塞任务运行中')
            # 加休眠避免占满CPU
            time.sleep(1)

class ThirdClass:
    def __init__(self):
        self.f_class = FirstClass()
        self.s_class = SecondClass()
    
    async def run_all(self):
        # 将同步阻塞方法包装为asyncio可调度的异步任务，自动放到独立线程运行
        sync_task = asyncio.to_thread(self.s_class.display_info)
        async_task = self.f_class.main()
        # 并发运行两个任务，任意一个报错都会正常抛出
        await asyncio.gather(sync_task, async_task)

if __name__ == "__main__":
    runner = ThirdClass()
    asyncio.run(runner.run_all())

方案说明：

• asyncio.to_thread由事件循环自动管理线程池，不会阻塞主循环的异步任务调度
• 所有任务在同一个asyncio上下文下运行，后续新增异步任务、做任务间通信都更方便
• 不需要手动处理线程启动、回收逻辑

手动管理线程的实现方案 #

如果确实需要手动创建线程分别跑两类任务，注意不要跨线程共享事件循环，异步任务所在线程必须单独启动自己的事件循环：

import asyncio
import time
from threading import Thread

class FirstClass:
    async def main(self):
        print("Hello 异步任务启动")
        while True:
            print("异步任务运行中，距上次输出间隔20s")
            await asyncio.sleep(20)
    
    def run_async_task(self):
        # 子线程内单独启动事件循环，禁止跨线程共享loop对象
        asyncio.run(self.main())

class SecondClass:
    def display_info(self):
        while True:
            print('People 同步阻塞任务运行中')
            time.sleep(1)

class ThirdClass:
    def __init__(self):
        self.f_class = FirstClass()
        self.s_class = SecondClass()
    
    def run_all(self):
        thread_list = [
            # target只传方法引用，不要加括号直接调用
            Thread(target=self.s_class.display_info, daemon=True),
            Thread(target=self.f_class.run_async_task, daemon=True)
        ]
        for t in thread_list:
            t.start()
        # 阻塞主线程等待子线程运行
        for t in thread_list:
            t.join()

if __name__ == "__main__":
    runner = ThirdClass()
    runner.run_all()

注意事项：

• 给Thread传target时只传方法/函数名，不要加()直接调用，否则会在主线程同步执行
• asyncio事件循环本身不是线程安全的，绝对不要跨线程传递、调用其他线程创建的loop对象
• 手动创建线程建议设置daemon=True，避免主线程退出后子线程变成孤儿进程残留
• 无限循环的同步任务必须加合理的休眠间隔，避免CPU资源被占满

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