在 Python 中使用多线程实现 Web 爬虫时，限速控制是保障爬虫稳定运行、避免对目标服务器造成过大压力的关键环节。以下是几种优雅实现限速控制的方法。 ### 3.1 使用 `time.sleep` 实现简单限速 最基础的限速方式是通过 `time.sleep()` 在每次请求之间添加延迟。虽然实现简单，但该方法缺乏灵活性，难以应对高并发场景。 ```python import time import requests from threading import Thread def fetch(url): response = requests.get(url) print(response.status_code) time.sleep(1) # 每次请求间隔1秒 urls = ["https://example.com"] * 10 threads = [Thread(target=fetch, args=(url,)) for url in urls] for t in threads: t.start() for t in threads: t.join() ``` 此方法适用于低并发场景，但不推荐用于大规模爬虫任务，因为它无法动态调整请求频率[^1]。 ### 3.2 使用 `threading.Semaphore` 控制并发数量 通过 `Semaphore` 可以限制同时运行的线程数量，从而间接控制请求频率。这种方式更适用于控制并发请求总数，避免服务器过载。 ```python from threading import Semaphore, Thread import requests semaphore = Semaphore(5) # 同时最多5个线程执行请求 def fetch(url): with semaphore: response = requests.get(url) print(response.status_code) urls = ["https://example.com"] * 20 threads = [Thread(target=fetch, args=(url,)) for url in urls] for t in threads: t.start() for t in threads: t.join() ``` 此方法可以有效控制并发数量，适用于需要统一限流的多线程爬虫场景[^1]。 ### 3.3 使用令牌桶算法进行动态限速 令牌桶算法是一种更高级的限速策略，它通过周期性地向桶中添加令牌，只有获取到令牌的请求才能执行，支持突发请求且控制更精细。 ```python import time from threading import Thread, Semaphore class TokenBucket: def __init__(self, rate): self.rate = rate self.tokens = rate self.last_time = time.time() self.lock = Semaphore() def consume(self, tokens=1): with self.lock: now = time.time() elapsed = now - self.last_time self.tokens += elapsed * self.rate self.last_time = now if self.tokens > self.rate: self.tokens = self.rate if self.tokens >= tokens: self.tokens -= tokens return True else: return False bucket = TokenBucket(2) # 每秒允许2个请求 def fetch(url): while not bucket.consume(): time.sleep(0.1) print(f"Fetching {url}") urls = ["https://example.com"] * 10 threads = [Thread(target=fetch, args=(url,)) for url in urls] for t in threads: t.start() for t in threads: t.join() ``` 该方法在多线程环境下能够实现更灵活、精确的限速控制，适用于高并发抓取任务。 ### 3.4 使用 `concurrent.futures.ThreadPoolExecutor` 配合限速机制 结合 `ThreadPoolExecutor` 与限速逻辑，可以更好地管理线程池并实现请求调度。 ```python from concurrent.futures import ThreadPoolExecutor import time import requests def fetch(url): response = requests.get(url) print(response.status_code) time.sleep(0.5) # 每次请求间隔0.5秒 urls = ["https://example.com"] * 10 with ThreadPoolExecutor(max_workers=5) as executor: executor.map(fetch, urls) ``` 此方法通过线程池控制并发数，并结合固定延迟实现限速，适用于需要兼顾性能与限速的场景[^5]。 --- ###