如何为Python异步回声服务器添加优雅关闭的信号处理？

现有异步回声服务器代码，需添加SIGINT（Ctrl+C）和SIGTERM信号处理实现优雅关闭——关闭所有连接的writer，同时兼容不在服务器监听范围内的其他异步任务，以下是具体实现：

完整实现代码 #

import asyncio
import signal
from asyncio import StreamReader, StreamWriter

# 全局关闭事件，用于通知所有异步任务触发退出逻辑
shutdown_event = asyncio.Event()

def handle_signal(signum, frame):
    """信号处理函数，收到终止信号时设置关闭事件"""
    print(f"\nReceived signal {signal.Signals(signum).name}, starting shutdown...")
    shutdown_event.set()

async def echo(reader: StreamReader, writer: StreamWriter): 
    peername = writer.get_extra_info('peername')
    connection_id = f"{peername[0]}:{peername[1]}"
    print(f'Open connection to {connection_id=}')

    try:
        while not shutdown_event.is_set():
            # 同时监听数据读取和关闭事件，避免阻塞在readline无法响应关闭信号
            done, pending = await asyncio.wait(
                [reader.readline(), shutdown_event.wait()],
                return_when=asyncio.FIRST_COMPLETED
            )
            # 处理客户端发送的数据
            if reader.readline() in done:
                data = done.pop().result()
                if not data:
                    break  # 客户端主动断开连接
                print(f'Received data {data=}')
                data = data.decode()
                writer.write(f'Return as uppercase:{data.upper()}'.encode())
                await writer.drain()
            # 响应关闭信号，退出循环
            if shutdown_event.wait() in done:
                break
    finally:
        # 确保连接资源被释放
        print(f'Closing connection to {connection_id=}')
        writer.close()
        await writer.wait_closed()

async def other_async_task():
    """示例独立异步任务，遵循统一的优雅退出逻辑"""
    while not shutdown_event.is_set():
        print("Other async task is running...")
        await asyncio.sleep(2)
    print("Other async task exiting gracefully.")

async def main(host='127.0.0.1', port=7000):
    # 注册信号处理器
    loop = asyncio.get_running_loop()
    for sig in (signal.SIGINT, signal.SIGTERM):
        loop.add_signal_handler(sig, handle_signal, sig, None)

    print(f'Server running on {host}:{port}')
    # 启动回声服务器
    server = await asyncio.start_server(echo, host, port)
    # 启动其他异步任务
    other_task = asyncio.create_task(other_async_task())

    try:
        # 等待关闭信号触发
        await shutdown_event.wait()
        # 停止服务器接受新连接
        server.close()
        await server.wait_closed()
        print("Server stopped accepting new connections.")
        # 等待所有异步任务完成退出
        await asyncio.gather(other_task, return_exceptions=True)
    finally:
        print("Shutdown completed.")

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

关键实现说明 #

• 统一关闭信号：用asyncio.Event()作为全局关闭标记，所有异步任务通过检查事件状态实现同步退出，避免分散的关闭逻辑。
• 信号处理适配：通过loop.add_signal_handler注册SIGINT和SIGTERM信号的同步处理函数，确保信号能触发全局关闭事件。
• 连接优雅关闭：echo函数不再单纯阻塞在reader.readline()，而是通过asyncio.wait同时监听数据读取和关闭事件，收到信号后立即退出循环，并在finally块中确保writer被正确关闭。
• 兼容其他异步任务：所有独立异步任务只需在循环中检查shutdown_event.is_set()，即可遵循统一的优雅退出逻辑，无需修改核心关闭流程。
• 完整关闭流程：触发关闭后，先停止服务器接受新连接，再等待现有连接处理完成，最后等待所有异步任务退出，确保服务无残留资源泄漏。

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