如何在Python中创建可终止的后台任务？

看起来你遇到的核心问题是同步阻塞的循环卡住了asyncio事件循环，导致任务根本没机会响应取消操作。让我一步步拆解问题和解决方案：

问题本质 #

asyncio的任务取消机制依赖于任务在执行过程中暂停并让出事件循环控制权（也就是await某个可等待对象的时候）。你的data_processor里的for item in source是纯同步的循环——一旦这个循环开始，它会霸占事件循环的所有时间片，完全不给其他操作（包括取消信号的处理）留机会。所以你调用processing_task.cancel()后，任务根本没法接收到CancelledError，自然不会终止。

针对性解决方案 #

根据你的create_data_source是否可修改，有几种不同的处理方式：

方案1：将同步生成器改为异步生成器（优先推荐，如果可控） #

如果create_data_source是你自己实现的，或者可以修改为异步逻辑，直接用async for迭代异步生成器，这样每次迭代都会让出事件循环控制权，取消信号就能正常触发了。

from typing import AsyncGenerator
import asyncio

# 改为异步生成器
async def create_data_source() -> AsyncGenerator[int, None]:
    i = 0
    while True:
        yield i
        i += 1
        await asyncio.sleep(0.1)  # 模拟异步数据生成，强制让出控制权

async def data_processor(source: AsyncGenerator[int, None]) -> None:
    try:
        async for item in source:
            print("Working")
    except asyncio.CancelledError:
        print("Cancelled via exception")
        raise  # 必须重新抛出，让asyncio正确处理任务取消状态
    except Exception as e:
        print("Exception", e)
        raise
    finally:
        print("Cleanup completed")

async def main():
    data_source = create_data_source()
    processing_task = asyncio.create_task(data_processor(data_source))

    await asyncio.sleep(3)
    processing_task.cancel()
    # 等待任务处理取消并完成清理
    try:
        await processing_task
    except asyncio.CancelledError:
        pass

asyncio.run(main())

方案2：用线程池隔离同步逻辑（如果生成器不可修改） #

如果create_data_source是第三方库返回的同步生成器，没法改成异步，就把同步循环放到线程池里，用asyncio.Event作为跨线程的终止信号——这样asyncio事件循环不会被阻塞，能正常处理终止信号。

from typing import Generator, Any
import asyncio

# 假设这是你无法修改的同步生成器
def create_data_source() -> Generator[int, Any, None]:
    i = 0
    while True:
        yield i
        i += 1

async def data_processor(source: Generator[int, Any, None], shutdown_event: asyncio.Event) -> None:
    def sync_processing_loop():
        try:
            for item in source:
                # 每次迭代前检查终止信号
                if shutdown_event.is_set():
                    break
                print("Working")
                # 若为计算密集型任务，需偶尔让出线程控制权
                # import time
                # time.sleep(0)
        except Exception as e:
            print("Exception", e)
            raise
        finally:
            print("Cancelled and cleaned up")

    # 用asyncio.to_thread把同步逻辑丢到线程池（Python3.9+支持，你的3.11完全兼容）
    await asyncio.to_thread(sync_processing_loop)

async def main():
    data_source = create_data_source()
    shutdown_event = asyncio.Event()
    processing_task = asyncio.create_task(data_processor(data_source, shutdown_event))

    await asyncio.sleep(3)
    # 发送终止信号
    shutdown_event.set()
    # 等待线程任务完成清理
    await processing_task

asyncio.run(main())

注意：如果同步循环是极端密集的计算任务，需要在循环里偶尔加time.sleep(0)——这会强制Python的GIL切换线程，让shutdown_event.is_set()能及时生效。

方案3：在同步循环中强制让出控制权（临时妥协方案） #

如果不想改生成器也不想用线程，可以在同步循环中每隔几次迭代就调用await asyncio.sleep(0)——这个操作不会实际等待，但会强制让出事件循环控制权，让取消信号有机会被处理：

async def data_processor(source: Generator[int, Any, None]) -> None:
    try:
        for idx, item in enumerate(source):
            print("Working")
            # 每10次迭代让出一次控制权
            if idx % 10 == 0:
                await asyncio.sleep(0)
    except asyncio.CancelledError:
        print("Cancelled")
        raise
    except Exception as e:
        print("Exception", e)
        raise
    finally:
        print("Cleanup done")

这种方法的缺点是取消响应有延迟（要等下一次让出控制权的时机），适合迭代本身有一定耗时的场景。

总结 #

asyncio不是“能终止任何任务”的黑盒——它只能终止那些愿意让出事件循环控制权的异步任务。你的原始代码把纯同步的阻塞逻辑直接塞到asyncio任务里，导致事件循环被完全卡住，自然没法处理取消信号。选择哪种方案，核心看你是否能控制create_data_source的实现逻辑。

备注：内容来源于stack exchange，提问作者Docom