多线程运行用户名检测函数时如何让认领函数仅执行单次请求

原代码问题说明 #

原来的实现中检测线程直接调用认领接口，多线程并发检测时如果同时命中多个可用用户名，会瞬间触发大量认领请求，远超5次/秒的接口限流阈值，同时检测逻辑和认领逻辑耦合，认领的等待逻辑会拖慢检测的请求速率。

核心实现思路 #

将检测和认领逻辑完全解耦：检测线程只负责发送检测请求，命中可用用户名后仅将用户名写入队列，不直接调用认领接口；单独启动一个后台线程处理认领队列，严格控制认领请求的发送速率，完全不影响检测端的请求效率。

调整后完整可运行代码 #

import threading
import requests
import queue
import time
from concurrent.futures import ThreadPoolExecutor

class UsernameChecker:
    def __init__(self):
        # 可用用户名缓冲队列
        self.available_queue = queue.Queue()
        # 启动后台认领线程
        threading.Thread(target=self._claim_worker, daemon=True).start()

    # 后台认领工作线程，严格控制速率
    def _claim_worker(self):
        while True:
            user = self.available_queue.get()
            try:
                # 调用认领接口
                requests.post(f'https://somewebsite.com/{user}')
                # 可自行添加认领成功后的处理逻辑
            finally:
                self.available_queue.task_done()
            # 固定间隔0.2秒，刚好满足每秒最多5次的限流要求
            time.sleep(0.2)

    def claim(self, user):
        # 仅将待认领用户名写入队列，无额外阻塞
        self.available_queue.put(user)

    def check(self, name):
        # 检测用户名可用性，无额外阻塞
        resp = requests.get(f'https://somewebsite.com/{name}')
        if 'available' in resp.text:
            self.claim(name)

    def run(self, check_name_list, max_check_threads=150):
        # 用线程池统一管理检测线程，调整max_workers参数即可匹配200rq/s的速率要求
        with ThreadPoolExecutor(max_workers=max_check_threads) as executor:
            executor.map(self.check, check_name_list)

# 使用示例
if __name__ == "__main__":
    # 替换为你要检测的用户名列表
    target_names = [f"test{i}" for i in range(10000)]
    checker = UsernameChecker()
    checker.run(target_names)

实现说明 #

• 速率控制：认领逻辑单独在后台线程运行，每秒最多发送5次请求，就算同时有大量可用用户名被检测到，也会在队列内排队按顺序发送，不会触发限流。
• 检测速率保障：检测逻辑无任何额外等待逻辑，仅需调整max_check_threads参数即可轻松达到200rq/s的检测速率要求，完全不受认领逻辑影响。
• 稳定性优化：使用线程池管理检测线程，避免手动创建大量线程导致的系统资源开销过高，原代码中存在的check方法未接收name参数、线程启动时方法名拼写错误的问题已在上述代码中修正。

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