效率炸裂！Python 多线程爬虫实现 10 倍速采集

import requests
from bs4 import BeautifulSoup
import time

# 待爬取的URL列表（模拟100个目标链接）
def generate_urls():
    base_url = "https://example-blog.com/article/{}"  # 替换为真实测试链接
    return [base_url.format(i) for i in range(1, 101)]

# 单线程爬取函数
def single_thread_crawl(urls):
    start_time = time.time()
    results = []
    for url in urls:
        try:
            # 发起请求（模拟网络等待）
            response = requests.get(url, timeout=5)
            response.raise_for_status()  # 抛出HTTP错误
            soup = BeautifulSoup(response.text, 'html.parser')
            # 提取文章标题（根据目标网站结构调整）
            title = soup.find('h1', class_='article-title').text.strip()
            results.append(title)
            print(f"单线程已爬取：{title}")
        except Exception as e:
            print(f"单线程爬取{url}失败：{e}")
            results.append(None)
    end_time = time.time()
    print(f"\n单线程爬取完成，总耗时：{end_time - start_time:.2f}秒")
    return results

if __name__ == "__main__":
    urls = generate_urls()
    single_thread_crawl(urls)

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

# 全局队列：存储待爬取的URL
url_queue = Queue()
# 全局列表：存储爬取结果（需加锁保证线程安全）
results = []
lock = threading.Lock()

# 爬取任务函数（每个线程执行的逻辑）
def crawl_worker():
    while not url_queue.empty():
        url = url_queue.get()  # 从队列获取任务
        try:
            response = requests.get(url, timeout=5)
            response.raise_for_status()
            soup = BeautifulSoup(response.text, 'html.parser')
            title = soup.find('h1', class_='article-title').text.strip()
            # 加锁写入结果，避免多线程冲突
            with lock:
                results.append(title)
                print(f"线程{threading.current_thread().name}已爬取：{title}")
        except Exception as e:
            with lock:
                print(f"线程{threading.current_thread().name}爬取{url}失败：{e}")
                results.append(None)
        finally:
            url_queue.task_done()  # 标记任务完成

# 多线程爬取主函数
def multi_thread_crawl(urls, thread_num=10):
    start_time = time.time()
    # 将URL填入队列
    for url in urls:
        url_queue.put(url)
    # 创建并启动线程
    threads = []
    for i in range(thread_num):
        t = threading.Thread(name=f"Thread-{i+1}", target=crawl_worker)
        t.start()
        threads.append(t)
    # 等待队列所有任务完成
    url_queue.join()
    # 等待所有线程结束
    for t in threads:
        t.join()
    end_time = time.time()
    print(f"\n多线程爬取完成，线程数：{thread_num}，总耗时：{end_time - start_time:.2f}秒")
    return results

if __name__ == "__main__":
    urls = generate_urls()  # 复用之前的URL生成函数
    multi_thread_crawl(urls, thread_num=10)

# 优化后的爬取函数（添加重试）
def crawl_worker_with_retry(max_retry=3):
    while not url_queue.empty():
        url = url_queue.get()
        retry_count = 0
        while retry_count < max_retry:
            try:
                response = requests.get(url, timeout=5)
                response.raise_for_status()
                # 解析逻辑（同上）
                break  # 成功则退出重试循环
            except Exception as e:
                retry_count += 1
                if retry_count == max_retry:
                    with lock:
                        print(f"线程{threading.current_thread().name}爬取{url}重试{max_retry}次失败：{e}")
                time.sleep(0.5)  # 重试间隔
        url_queue.task_done()