如何优雅停止含异步任务的Python服务？求更优实现方案

问题描述 #

我开发了一个小型Python服务，通过asyncio运行多个无限循环的异步任务，用于检查K8s Pod状态并将计算结果存入数据库。我希望实现服务的优雅停止/重启，即终止服务时让所有运行中的任务完成。我已编写了一段代码，但认为其实现不够优雅，尤其是全局变量的使用和main函数中的两个while循环。请问在Python 3.10环境下，是否有更高效简洁的实现方式？

原代码：

import asyncio
import signal
    
running = True
tasks = set()
    
async def sometask():
    global running
    while running:
        print('Some work on internet')
        await asyncio.sleep(5)
        print('Some work part 2')

def shutdown():
    global running
    print('shutdown')
    running = False

async def main():      
    loop = asyncio.get_running_loop()    
    loop.add_signal_handler(signal.SIGINT, shutdown)
    
    task: asyncio.Task = asyncio.create_task(sometask())
    task.add_done_callback(tasks.discard)
    tasks.add(task)
    
    while running:
        print(f'Running: {running}...{len(tasks)}')
        await asyncio.sleep(1)
        
    while len(tasks) > 0:
        print(f'Waiting for {len(tasks)} tasks to finish')
        await asyncio.sleep(1)
        
    print('Finish')
    

asyncio.run(main(), debug=True)

优化方案 #

Python 3.10+可以利用asyncio.Event替代全局变量，结合asyncio.gather和异步信号处理逻辑，彻底消除全局变量和轮询式while循环，实现更简洁高效的优雅停止：

优化后代码 #

import asyncio
import signal

async def sometask(stop_event: asyncio.Event):
    while not stop_event.is_set():
        print('Some work on internet')
        # 用wait_for包裹事件等待，支持收到停止信号时立即唤醒
        try:
            await asyncio.wait_for(stop_event.wait(), timeout=5)
        except asyncio.TimeoutError:
            # 超时说明无停止信号，执行后续任务逻辑
            print('Some work part 2')

async def shutdown(stop_event: asyncio.Event, tasks: set[asyncio.Task]):
    print('shutdown')
    stop_event.set()
    # 等待所有任务完成，return_exceptions避免单个任务异常导致整体失败
    await asyncio.gather(*tasks, return_exceptions=True)

async def main():
    stop_event = asyncio.Event()
    tasks = set()
    
    # 创建异步任务并加入集合管理
    task = asyncio.create_task(sometask(stop_event))
    task.add_done_callback(tasks.discard)
    tasks.add(task)
    
    # 注册信号处理：将同步信号转为异步停止任务
    loop = asyncio.get_running_loop()
    for sig in [signal.SIGINT, signal.SIGTERM]:
        loop.add_signal_handler(
            sig,
            lambda s=sig: asyncio.create_task(shutdown(stop_event, tasks))
        )
    
    # 等待所有任务自然结束
    await asyncio.gather(*tasks)
    print('Finish')

asyncio.run(main(), debug=True)

核心优化点 #

• 移除全局变量：用asyncio.Event作为停止信号载体，通过事件状态控制任务循环，避免全局变量的耦合问题。
• 消除轮询逻辑：
  • 任务内部用asyncio.wait_for替代固定sleep，收到停止信号时任务可立即唤醒，无需等待sleep周期结束，响应更及时。
  • main函数直接通过asyncio.gather等待所有任务完成，无需轮询检查任务数量。
• 异步信号处理：将信号触发的停止逻辑转为异步任务，在其中统一设置停止事件并等待任务收尾，逻辑更清晰。
• 覆盖多停止场景：同时处理SIGINT（Ctrl+C）和SIGTERM（系统终止信号），适配更多服务停止场景。

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