如何阻止Python在后台异步任务运行时终止程序？

你的核心问题是后台异步任务随主线程结束被强制终止，以及同步输入操作阻塞事件循环，以下是针对性的解决方法：

1. 用asyncio.gather管理多任务生命周期 #

原来的代码中，你只等待了loop_test的线程任务，该任务2秒完成后asyncio.run()就会退出，导致后台流式任务被直接终止。正确的做法是让事件循环同时等待后台任务和交互任务，直到两者都完成。

2. 用asyncio.to_thread实现非阻塞交互 #

input()是同步阻塞操作，直接放在asyncio事件循环里会卡住所有异步任务。用asyncio.to_thread把输入操作放到单独线程执行，既能获取用户输入，又不影响后台流式任务的运行。

修改后的完整代码示例 #

import asyncio
import time
from pprint import pprint

def loop_test():
    print("task is running")
    time.sleep(2)
    print("task is finished")

class TDA_Streaming:
    def __init__(self):
        self.streaming = False
        # 假设td_stream_client已完成初始化
        self.td_stream_client = None

    async def start_pipeline(self):
        print("Starting TD Stream")

        # 构建数据管道
        await self.td_stream_client.build_pipeline()

        data_response_count = 0
        self.streaming = True

        # 持续接收流式数据
        while self.streaming:
            print("Streaming")
            data = await self.td_stream_client.start_pipeline()

            # 解析数据内容
            if 'data' in data:
                content = data['data'][0]['content']

                print("Key: {}".format(content[0]['key']))
                pprint.pprint(content, indent=4)
                print('-' * 80)

                data_response_count += 1
        print("Done with while loop")

async def main():
    # 初始化流式客户端实例
    td_stream_client = TDA_Streaming()
    
    # 创建后台流式任务
    stream_task = asyncio.create_task(td_stream_client.start_pipeline())
    
    # 定义非阻塞交互任务
    async def interactive_task():
        # 先执行示例任务loop_test
        await asyncio.to_thread(loop_test)
        # 进入用户输入循环
        while True:
            user_input = await asyncio.to_thread(input, "\n输入命令（exit=停止流式任务）：")
            if user_input.lower() == "exit":
                td_stream_client.streaming = False
                print("正在停止流式任务...")
                break
            print(f"执行命令：{user_input}")
    
    # 同时运行两个任务，直到全部完成
    await asyncio.gather(stream_task, interactive_task())

asyncio.run(main())

关键细节说明 #

• asyncio.gather(stream_task, interactive_task())会等待两个任务都完成才结束程序，只要交互任务不退出，后台流式任务就会持续运行。
• 当用户输入exit，设置streaming=False，后台任务的while循环会自然退出，避免强制终止带来的资源泄漏。
• 若使用Python 3.9以下版本，可替换asyncio.to_thread为loop.run_in_executor(None, input, ...)实现相同效果。

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