Python爬虫线程处理完所有链接后如何停止运行？

你的代码目前存在几个导致无法停止的核心问题：重复读取文件引发重复处理、主逻辑无限循环启动线程、缺少线程终止的判断条件。以下是针对性的解决方案：

核心优化思路 #

• 一次性读取所有待处理链接，避免重复IO操作
• 使用线程安全的队列管理任务，每个线程从队列取任务，队列为空时自动退出
• 主程序等待所有工作线程完成后，自动终止整个程序

修正后的完整代码 #

import time
import os
import threading
import random
import requests
from bs4 import BeautifulSoup
from queue import Queue

# 全局统计变量
reqs = 0
success = 0
fails = 0
rps = 0
rpm = 0
_lock = threading.Lock()

def rpsm_loop():
    global rps, rpm, reqs
    while True:
        initial = reqs
        time.sleep(1)
        # 修正RPS计算：每秒请求数 = 1秒内的请求增量
        rps = round(reqs - initial, 1)
        rpm = round(rps * 60, 1)
        # 当所有任务完成且无活跃工作线程时，退出统计线程
        if queue.empty() and threading.active_count() <= 2:  # 2对应主线程+统计线程
            break

def scrapeEmails(queue):
    global reqs, success, fails
    while not queue.empty():
        url = queue.get()
        try:
            html_url = url.strip() + "/about"
            # 发送请求
            response = requests.get(
                html_url,
                cookies={'CONSENT': f'YES+cb.20221228-17-p0.en-GB+FX+{random.randint(100, 999)}'},
                # proxies=proxies  # 需要代理时取消注释
            )
            response.raise_for_status()  # 捕获HTTP错误
            
            soup = BeautifulSoup(response.text, "html.parser")
            # 此处执行查找邮箱的逻辑，示例：
            emails = soup.find_all(string=lambda text: "@" in text if text else False)
            if emails:
                with _lock:
                    success +=1
                    # 写入邮箱到文件
                    with open("emails.txt", "a", encoding="utf-8") as f:
                        f.write(f"{url}: {', '.join(emails)}\n")
            
            with _lock:
                reqs +=1
        except Exception as e:
            with _lock:
                fails +=1
                print(f"处理{url}失败: {str(e)}")
        finally:
            queue.task_done()

if __name__ == "__main__":
    # 1. 一次性读取所有链接并去重
    links_path = os.path.join("channel_links.txt")
    with open(links_path, "r", encoding="utf-8") as f:
        all_links = list(set(line.strip() for line in f if line.strip()))  # 去重
    
    # 2. 初始化任务队列
    queue = Queue()
    for link in all_links:
        queue.put(link)
    
    # 3. 启动统计线程
    threading.Thread(target=rpsm_loop, daemon=True).start()
    
    # 4. 启动工作线程（最多9个，加上主线程和统计线程不超过11）
    max_workers = 9
    for _ in range(max_workers):
        threading.Thread(target=scrapeEmails, args=(queue,), daemon=True).start()
    
    # 5. 等待所有任务完成
    queue.join()
    
    # 6. 输出最终统计结果
    print(f"任务完成！总请求数: {reqs}, 成功: {success}, 失败: {fails}, 最后RPS: {rps}, RPM: {rpm}")

关键修改说明 #

1. 任务队列管理：用queue.Queue存储所有链接，每个线程从队列取任务，处理完标记任务完成，队列为空时线程自动退出。
2. 一次性读取链接：只读取一次文件并去重，避免重复处理相同链接，提升效率。
3. 线程终止条件：主程序调用queue.join()等待所有任务完成，之后自动退出；统计线程在队列为空且无工作线程时也会停止。
4. 线程安全的统计：所有全局统计变量的修改都加了锁_lock，避免多线程竞争导致数据错误。
5. 修正RPS计算：原来的RPS计算逻辑错误，现在改为每秒计算一次请求增量，符合实际含义。

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