如何应用多线程提升BeautifulSoup爬取数据的速度

你的爬取任务属于典型的IO密集型场景，绝大多数运行时间都消耗在等待HTTP请求响应上，单线程逐个请求的模式会把大量时间浪费在空等上，用多线程并发处理请求可以把这部分等待时间利用起来，爬取速度会有非常明显的提升。

改造思路 #

• 直接用Python标准库自带的concurrent.futures.ThreadPoolExecutor实现多线程，不需要安装第三方依赖，上手门槛低，足够应对当前爬取场景
• 把单个详情页的爬取逻辑抽成独立函数，交给线程池并发执行
• 控制合理的线程数量，避免并发过高给目标站点造成压力、导致自身IP被封禁，一般设置5-10个线程就能获得数倍的速度提升
• 顺手修复原代码里的两个小问题：一是元素查找失败时变量未定义就引用的bug，二是select方法传参错误导致元素选取不准的问题

改造后完整代码 #

import requests
from bs4 import BeautifulSoup
import pandas as pd
from concurrent.futures import ThreadPoolExecutor, as_completed


url = "https://baroul-timis.ro/get-av-data?param=toti-avocatii"
base_url = 'https://baroul-timis.ro'
headers = {
    'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.114 Safari/537.36'
}

# 先获取所有律师详情页链接
productlink = []
data = requests.get(url, headers=headers, timeout=10).json()
for i, d in enumerate(data["data"], 1):
    link = BeautifulSoup(d["actions"], "html.parser").a["href"]
    comp = base_url + link
    productlink.append(comp)

# 封装单页爬取逻辑
def crawl_detail(link):
    res = {}
    try:
        r = requests.get(link, headers=headers, timeout=10)
        soup = BeautifulSoup(r.content, 'html.parser')
        user_info = soup.find_all('div', class_='user-info text-left mb-50')
        for info_block in user_info:
            res['title'] = info_block.find('h4').text.strip()
            # 修复原select传参错误问题，用find_all精准匹配带对应class的span
            span_items = info_block.find_all('span', class_='font-weight-bolder')
            # 捕获索引不存在的异常，避免变量未定义
            try:
                res['phone'] = span_items[2].text.split('\xa0')
            except (IndexError, AttributeError):
                res['phone'] = None
            try:
                res['email'] = span_items[3].text.split('\xa0')
            except (IndexError, AttributeError):
                res['email'] = None
        return res
    except Exception as e:
        print(f"爬取{link}失败：{str(e)}")
        return None

# 多线程并发爬取
test = []
# 线程数设为8，可根据实际网络情况调整，不建议超过15
with ThreadPoolExecutor(max_workers=8) as executor:
    task_list = [executor.submit(crawl_detail, link) for link in productlink]
    for task in as_completed(task_list):
        result = task.result()
        if result:
            test.append(result)

df = pd.DataFrame(test)
print(df)

注意事项 #

• 线程数不要盲目调高，并发过高不仅容易被站点封禁IP，也会占用你本地过多的网络资源
• 所有网络请求都加了timeout=10参数，避免个别无响应的请求卡死整个爬取任务
• 如果后续需要进一步提升稳定性，可以在请求之间增加随机延时、搭配代理IP池使用

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