如何用Python实现多线程批量Post请求以提升调用效率？

原来的单循环逐个发请求确实慢得让人着急，咱们用Python的concurrent.futures.ThreadPoolExecutor来实现多线程批量请求，既能提升速度，还能方便管理线程。不过要注意两个关键点：线程安全的文件写入（不然多个线程同时写文件会乱码或者丢内容），还有控制线程数量（别一下子发太多请求被服务器拉黑）。

改造思路 #

• 把单个Apple ID的验证逻辑封装成独立函数，方便线程调用
• 使用线程池批量提交任务，控制并发数
• 给文件操作加锁，保证多线程下写入不冲突

改造后的完整代码 #

import requests
from concurrent.futures import ThreadPoolExecutor
import threading

# 配置部分保持不变
env = "PROD"
accNameLocked = env
accountPassword = "ddd"
appIdKey = "3b356c1bac5ad9735ad62f24d43414eb59715cc4d21b178835626ce0d2daa77d"
fdcBrowserData = ""
language = "US-EN"
openiForgotInNewWindow = "true"
path = "/"
requestUri = "/login.html"
rv = "1"
scnt = "31b0020ec593370dc388f8965e"
view = "5"
url = "这里要填你的请求URL哦"  # 原代码里缺失url，记得补上实际地址！
appleId_file = "你的Apple ID文件路径"  # 替换成实际的文件路径

# 创建文件锁，解决多线程写入冲突
file_lock = threading.Lock()

def verify_apple_id(apple_id):
    apple_id = apple_id.strip()
    if not apple_id:  # 跳过空行
        return
    
    get_data = {
        "accountPassword": accountPassword,
        "appIdKey": appIdKey,
        "appleId": apple_id,
        "fdcBrowserData": fdcBrowserData,
        "requestUri": requestUri,
        "rv": rv,
        "scnt": scnt,
        "view": view
    }
    
    try:
        # 给请求加超时时间，避免线程无限挂起
        response = requests.post(url, params=get_data, timeout=10)
        response_content = response.content.decode('utf-8')  # 转字符串方便关键词判断
        
        valid_msg = "YAY"
        dia_msg = "Hey"
        
        if valid_msg in response_content:
            print(f"Valid :>> {apple_id}")
            with file_lock:  # 加锁确保单线程写入
                with open("rzlt/live.txt", "a+", encoding='utf-8') as f:
                    f.write(f"\n{apple_id}")
        elif dia_msg in response_content:
            print(f"Die :>> {apple_id}")
            with file_lock:
                with open("rzlt/die.txt", "a+", encoding='utf-8') as f:
                    f.write(f"\n{apple_id}")
        else:
            print(f"Unknown response for :>> {apple_id}")
    except Exception as e:
        print(f"Error processing {apple_id}: {str(e)}")

if __name__ == "__main__":
    # 读取Apple ID列表
    with open(appleId_file, "r+", encoding='utf-8') as open_file:
        read_me = open_file.readlines()
    
    # 设置线程池大小，建议根据服务器承受能力调整（比如10-20）
    with ThreadPoolExecutor(max_workers=15) as executor:
        executor.map(verify_apple_id, read_me)

关键说明 #

1. 线程池的使用：ThreadPoolExecutor会自动管理线程的创建和销毁，max_workers控制并发数，别设太大，不然容易触发服务器反爬或导致请求失败。
2. 文件锁：用threading.Lock()创建锁，每次写入文件时用with file_lock:确保同一时间只有一个线程在写，避免内容混乱。
3. 异常处理：给请求加了超时时间，还捕获了全局异常，避免单个线程出错导致整个程序崩溃。
4. 字符串解码：把响应的bytes内容转成字符串，比直接操作bytes更方便判断关键词。

内容的提问来源于stack exchange，提问作者Cahya Darma Wijaya