Python执行长任务时加载动画实现及异步线程适配问题

核心原理说明 #

同步函数（比如PyGithub的所有方法）是阻塞式的——一旦调用，会占用当前线程直到任务完成，期间线程无法处理其他逻辑。要实现它和加载动画并行，必须把同步任务和动画放到不同的执行单元（线程/异步任务），并通过信号量控制动画的启停。

方案1：基于线程的实现（适合普通同步代码场景） #

用threading模块将同步耗时任务和加载动画分别放到独立线程，通过threading.Event作为停止信号。

import threading
import time
from github import Github

def loading_animation(stop_event):
    """循环打印加载动画，直到收到停止信号"""
    chars = "/-\\|"
    idx = 0
    while not stop_event.is_set():
        print(f"\rLoading... {chars[idx]}", end="", flush=True)
        idx = (idx + 1) % len(chars)
        time.sleep(0.1)
    print("\rLoading done!        ")  # 清除动画痕迹

def sync_github_task():
    """模拟PyGithub的同步耗时操作"""
    g = Github("your_access_token")
    repo = g.get_repo("octocat/Hello-World")
    open_issues = repo.get_issues(state="open")
    return [issue.title for issue in open_issues]

def main():
    # 创建停止信号事件
    stop_animation = threading.Event()
    
    # 启动动画线程
    animation_thread = threading.Thread(target=loading_animation, args=(stop_animation,))
    animation_thread.start()
    
    # 启动同步任务线程
    task_result = None
    def run_task():
        nonlocal task_result
        task_result = sync_github_task()
    
    task_thread = threading.Thread(target=run_task)
    task_thread.start()
    
    # 等待任务完成，停止动画
    task_thread.join()
    stop_animation.set()
    animation_thread.join()
    
    # 输出结果
    print(f"\nFetched open issues: {task_result}")

if __name__ == "__main__":
    main()

关键细节 #

• 同步任务在独立线程执行，不会阻塞动画线程的运行
• threading.Event是线程安全的信号量，用来通知动画线程停止
• flush=True确保打印的动画实时刷新，不会被缓冲

方案2：基于asyncio的异步实现（适合已有异步框架的场景） #

如果你的项目已经用了asyncio，可以用run_in_executor把同步函数包装成可await的异步任务，避免阻塞事件循环。

import asyncio
import time
from github import Github

async def loading_animation(stop_event):
    """异步加载动画，通过await让出事件循环"""
    chars = "/-\\|"
    idx = 0
    while not stop_event.is_set():
        print(f"\rLoading... {chars[idx]}", end="", flush=True)
        idx = (idx + 1) % len(chars)
        await asyncio.sleep(0.1)  # 必须用await，否则会阻塞事件循环
    print("\rLoading done!        ")

def sync_github_task():
    """PyGithub同步任务，无async/await"""
    g = Github("your_access_token")
    repo = g.get_repo("octocat/Hello-World")
    open_issues = repo.get_issues(state="open")
    return [issue.title for issue in open_issues]

async def main():
    stop_animation = asyncio.Event()
    
    # 创建并启动动画异步任务
    animation_task = asyncio.create_task(loading_animation(stop_animation))
    
    # 用线程池执行同步任务，不阻塞事件循环
    loop = asyncio.get_running_loop()
    task_result = await loop.run_in_executor(None, sync_github_task)
    
    # 任务完成，停止动画
    stop_animation.set()
    await animation_task
    
    print(f"\nFetched open issues: {task_result}")

if __name__ == "__main__":
    asyncio.run(main())

关键细节 #

• 同步函数不能直接在异步事件循环中调用，否则会卡住整个循环
• run_in_executor将同步任务放到默认线程池执行，事件循环可以同时处理动画任务
• asyncio.Event是异步安全的信号量，用来通知动画任务停止

常见错误排查 #

1. 直接在主线程/事件循环线程调用同步函数：会导致线程被阻塞，动画无法执行
2. 没有用线程安全/异步安全的信号量：可能出现动画无法正确停止的情况
3. 动画循环没有让出执行权：比如异步方案中用time.sleep代替await asyncio.sleep，会阻塞事件循环

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