如何终止并退出由exec()调用的异步脚本？

问题场景 #

我有一个bs4_scraper.py文件，其中的bs4_scraper是异步函数，单独在终端执行python bs4_scraper.py时能正常完成并优雅退出。但在Scrapy的WritePidErrorsPipeline的close_spider方法中，通过exec(open(bs4_file).read())调用该脚本后，任务虽能运行完成，但进程会僵死无法自动退出。

相关代码 #

bs4_scraper.py

async def bs4_scraper():
    limit = Semaphore(8)
    tasks = []
    pids = get_pids_from_file_generator()
    for pid in pids:
        task = create_task(scrap(pid, fake_header(), limit))
        tasks.append(task)
    result = await gather(*tasks)
    return result


if __name__ == "__main__":
    try:
        run(bs4_scraper())
    except Exception as e:
        logger.error(e)

Scrapy管道代码

class WritePidErrorsPipeline:
    def close_spider(self, spider):
        pid_errors_file = generate_pid_errors_file()
        pg = PostgresDB()
        non_inserted_ids = pg.select_non_inserted_ids(pid_errors_file)
        if non_inserted_ids:
            self.insertion_errors_file(non_inserted_ids)
            bs4_file = os.path.abspath("bs4/bs4_scraper.py")
            exec(open(bs4_file).read()) # 问题所在
        else:
            logger.info("[SUCCESS]: There are no items missing")

    def insertion_errors_file(
        self,
        non_inserted_ids: List[Tuple[str]],
        output_file: str = "insertion_errors.log",
    ) -> str:
        with open(output_file, "w", encoding="utf-8") as f:
            for non_inserted_id in non_inserted_ids:
                f.write(f"{non_inserted_id[0]}\n")
        return output_file

补充说明：scrap是发起异步请求的异步函数，get_pids_from_file_generator是读取txt文件生成pid的生成器函数。

解决方案 #

1. 重构异步脚本为可导入模块 #

修改bs4_scraper.py，将异步执行逻辑封装成独立的同步调用函数，方便外部模块导入使用：

import asyncio
from asyncio import Semaphore, create_task, gather

# 原异步函数保持不变
async def bs4_scraper():
    limit = Semaphore(8)
    tasks = []
    pids = get_pids_from_file_generator()
    for pid in pids:
        task = create_task(scrap(pid, fake_header(), limit))
        tasks.append(task)
    result = await gather(*tasks)
    return result

# 封装同步调用函数，负责管理事件循环
def run_scraper():
    try:
        asyncio.run(bs4_scraper())
    except Exception as e:
        logger.error(e)

# 保留原终端执行入口
if __name__ == "__main__":
    run_scraper()

2. 修改Scrapy管道调用方式 #

替换exec的调用方式，改为导入模块并调用封装好的函数，确保事件循环正确关闭：

import os
import sys
# 将bs4目录添加到Python路径，确保能导入模块
sys.path.append(os.path.abspath("bs4"))
from bs4_scraper import run_scraper

class WritePidErrorsPipeline:
    def close_spider(self, spider):
        pid_errors_file = generate_pid_errors_file()
        pg = PostgresDB()
        non_inserted_ids = pg.select_non_inserted_ids(pid_errors_file)
        if non_inserted_ids:
            self.insertion_errors_file(non_inserted_ids)
            # 替换exec为函数调用
            run_scraper()
        else:
            logger.info("[SUCCESS]: There are no items missing")

    # insertion_errors_file方法保持不变
    def insertion_errors_file(
        self,
        non_inserted_ids: List[Tuple[str]],
        output_file: str = "insertion_errors.log",
    ) -> str:
        with open(output_file, "w", encoding="utf-8") as f:
            for non_inserted_id in non_inserted_ids:
                f.write(f"{non_inserted_id[0]}\n")
        return output_file

原因说明 #

• 使用exec(open(bs4_file).read())会让异步代码运行在Scrapy进程的上下文环境中，异步事件循环的资源无法被正确回收，导致进程僵死。
• 改为导入模块并调用asyncio.run()封装的函数后，asyncio.run()会负责创建独立的事件循环，在任务完成后自动关闭循环并释放所有相关资源，确保进程能正常退出。

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