如何让Python网站检测函数并行运行？求Threading学习资源

嘿，我懂你这种逐个等请求完成的痛苦——50个链接串行跑确实慢到让人抓狂！咱们直接用Python的threading模块来并行处理这些请求，能把耗时砍到原来的几分之一，下面给你具体方案和学习要点。

基础Threading实现方案 #

首先把单个链接的检测逻辑抽成独立函数，然后为每个链接创建一个线程来执行：

import requests
import threading

def check_single_link(id_num):
    url = f"https://www.WEBSITE.com/ABCDE?id={id_num}"
    try:
        # 加timeout防止请求卡住拖慢整体
        response = requests.get(url, allow_redirects=False, timeout=5)
        if response.status_code == 200:
            print(f"\033[92mWeb Exists!\033[0m {id_num}")
        else:
            print(f"\033[91mWeb Does Not Exist!\033[0m {id_num}")
    except requests.exceptions.RequestException as e:
        # 捕获请求异常（超时、连接失败等），避免单个线程崩溃影响全局
        print(f"\033[93mError checking {id_num}: {str(e)}\033[0m")

def checkExistingWeb():
    threads = []
    for i in range(1, 51):
        # 创建线程：target指定要执行的函数，args是传给函数的参数（注意是元组，单个参数要加逗号）
        thread = threading.Thread(target=check_single_link, args=(i,))
        threads.append(thread)
        thread.start()  # 启动线程
    
    # 等待所有线程执行完毕，避免主线程提前退出
    for thread in threads:
        thread.join()

if __name__ == "__main__":
    checkExistingWeb()

为什么这样更快？ #

原来的串行逻辑是等一个请求完全响应后再发下一个，大部分时间都在等服务器回复；并行是同时发起多个请求，把这些等待时间重叠起来，整体耗时会接近单个请求的最长响应时间，而不是所有请求时间的总和。

更简洁的进阶方案：ThreadPoolExecutor #

如果不想手动管理线程列表，可以用concurrent.futures.ThreadPoolExecutor，它帮你封装了线程池的管理逻辑，代码更简洁：

from concurrent.futures import ThreadPoolExecutor
import requests

def check_single_link(id_num):
    url = f"https://www.WEBSITE.com/ABCDE?id={id_num}"
    try:
        response = requests.get(url, allow_redirects=False, timeout=5)
        # 返回结果，后续统一处理输出
        return (id_num, True) if response.status_code == 200 else (id_num, False)
    except requests.exceptions.RequestException:
        # None表示请求出错
        return (id_num, None)

def checkExistingWeb():
    # max_workers控制并发数，建议根据目标网站的承受能力设置（10-20都可以，50个链接的话10足够）
    with ThreadPoolExecutor(max_workers=10) as executor:
        # map自动把range(1,51)的每个元素传给check_single_link
        results = executor.map(check_single_link, range(1, 51))
    
    # 统一处理结果输出
    for id_num, status in results:
        if status is True:
            print(f"\033[92mWeb Exists!\033[0m {id_num}")
        elif status is False:
            print(f"\033[91mWeb Does Not Exist!\033[0m {id_num}")
        else:
            print(f"\033[93mError checking {id_num}\033[0m")

if __name__ == "__main__":
    checkExistingWeb()

Threading模块核心学习要点 #

不用找外部资源，记住这几个核心点就够用：

• Thread类：创建线程的核心，target是要执行的函数，args是函数的参数（必须是元组，单个参数要加逗号(i,)）
• start()方法：启动线程，让函数在新的线程中执行（不是立即执行，而是等待系统调度）
• join()方法：阻塞主线程，直到该线程执行完毕，避免主线程先退出导致程序结束
• 线程安全：如果多个线程要修改同一个全局变量，需要用threading.Lock()加锁；但你这个场景只是打印，Python的print本身是线程安全的（只是输出顺序可能乱，不影响结果）
• 异常处理：每个线程内部的异常不会自动传递到主线程，所以一定要在任务函数里加try-except，不然某个线程报错会直接崩溃，其他线程可能还在运行

注意事项 #

• 不要设置过大的并发数：比如一下子开50个线程，可能会被目标网站判定为恶意请求而封IP，建议控制在10-20个以内
• 一定要加timeout：防止某个请求卡住（比如服务器无响应），拖慢整个程序的执行速度
• 可以考虑加重试机制：如果某个请求失败，自动重试1-2次（用tenacity库或者自己写循环）

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