当前位置：首页 > news >正文

Python多线程爬虫加速电商数据采集

news 2025/9/21 6:58:38

在当今的大数据时代，电商平台积累了海量的商品信息，这些数据对于市场分析、竞品研究和商业决策具有重要价值。然而，单线程爬虫在面对大量数据采集任务时效率低下，难以满足实际需求。本文将介绍如何使用 Python 多线程技术加速电商数据采集，并提供一个完整的实现示例。

多线程爬虫的优势

提高效率：多线程可以同时发起多个请求，充分利用网络带宽
节省时间：对于需要采集大量页面的任务，多线程能显著减少总耗时
资源合理利用：在等待一个请求响应的同时，可以处理其他请求

实现方案

我们将使用以下 Python 库实现多线程电商爬虫：

requests：发送 HTTP 请求获取网页内容
threading：实现多线程功能
queue：创建任务队列，实现线程间通信
BeautifulSoup：解析 HTML 页面，提取所需数据
csv：将采集的数据保存到 CSV 文件

完整代码示例

import requests
import threading
import queue
import time
import csv
from bs4 import BeautifulSoup
import random
import logging# 配置日志
logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(levelname)s - %(message)s',handlers=[logging.FileHandler('crawler.log'),logging.StreamHandler()]
)class EcommerceCrawler:def __init__(self, base_url, max_pages, num_threads=5, output_file='products.csv'):self.base_url = base_url  # 电商网站基础URLself.max_pages = max_pages  # 要爬取的最大页数self.num_threads = num_threads  # 线程数量self.output_file = output_file  # 输出文件self.task_queue = queue.Queue()  # 任务队列self.results = []  # 存储爬取结果self.lock = threading.Lock()  # 线程锁，用于安全地修改共享数据self.user_agents = ["Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36","Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.1.1 Safari/605.1.15","Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:89.0) Gecko/20100101 Firefox/89.0"]# 初始化任务队列for page in range(1, self.max_pages + 1):self.task_queue.put(page)def get_headers(self):"""生成随机请求头，模拟不同浏览器"""return {'User-Agent': random.choice(self.user_agents),'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8','Accept-Language': 'en-US,en;q=0.5','Connection': 'keep-alive','Upgrade-Insecure-Requests': '1'}def fetch_page(self, page_num):"""获取指定页面的内容"""try:url = f"{self.base_url}?page={page_num}"headers = self.get_headers()response = requests.get(url, headers=headers, timeout=10)# 检查响应状态if response.status_code == 200:logging.info(f"成功获取第 {page_num} 页内容")return response.textelse:logging.warning(f"获取第 {page_num} 页失败，状态码: {response.status_code}")return Noneexcept Exception as e:logging.error(f"获取第 {page_num} 页时发生错误: {str(e)}")return Nonedef parse_page(self, html):"""解析页面内容，提取商品信息"""try:soup = BeautifulSoup(html, 'html.parser')products = []# 这里的选择器需要根据实际电商网站的HTML结构进行调整product_items = soup.select('.product-item')  # 假设商品项的类名为product-itemfor item in product_items:# 提取商品信息，选择器需根据实际情况调整name_tag = item.select_one('.product-name')price_tag = item.select_one('.product-price')rating_tag = item.select_one('.product-rating')review_count_tag = item.select_one('.product-review-count')# 提取文本内容，处理可能的None值name = name_tag.get_text(strip=True) if name_tag else "N/A"price = price_tag.get_text(strip=True) if price_tag else "N/A"rating = rating_tag.get_text(strip=True) if rating_tag else "N/A"review_count = review_count_tag.get_text(strip=True) if review_count_tag else "N/A"product = {'name': name,'price': price,'rating': rating,'review_count': review_count}products.append(product)logging.info(f"成功解析 {len(products)} 个商品信息")return productsexcept Exception as e:logging.error(f"解析页面时发生错误: {str(e)}")return []def worker(self):"""线程工作函数，处理队列中的任务"""while not self.task_queue.empty():try:page_num = self.task_queue.get(block=False)html = self.fetch_page(page_num)if html:products = self.parse_page(html)# 使用锁确保线程安全地修改共享结果列表with self.lock:self.results.extend(products)# 标记任务完成self.task_queue.task_done()# 添加随机延迟，避免给服务器造成过大压力time.sleep(random.uniform(1, 3))except queue.Empty:breakexcept Exception as e:logging.error(f"线程工作时发生错误: {str(e)}")self.task_queue.task_done()def run(self):"""启动爬虫"""logging.info(f"开始爬取，总页数: {self.max_pages}, 线程数: {self.num_threads}")start_time = time.time()# 创建并启动线程threads = []for _ in range(self.num_threads):thread = threading.Thread(target=self.worker)thread.start()threads.append(thread)# 等待所有任务完成self.task_queue.join()# 等待所有线程结束for thread in threads:thread.join()# 保存结果self.save_results()end_time = time.time()elapsed_time = end_time - start_timelogging.info(f"爬取完成，共获取 {len(self.results)} 个商品信息，耗时: {elapsed_time:.2f} 秒")def save_results(self):"""将结果保存到CSV文件"""try:if not self.results:logging.warning("没有可保存的结果")return# 获取字段名fieldnames = self.results[0].keys()with open(self.output_file, 'w', newline='', encoding='utf-8') as csvfile:writer = csv.DictWriter(csvfile, fieldnames=fieldnames)writer.writeheader()writer.writerows(self.results)logging.info(f"结果已保存到 {self.output_file}")except Exception as e:logging.error(f"保存结果时发生错误: {str(e)}")if __name__ == "__main__":# 示例：爬取某电商网站的商品信息# 注意：请将URL替换为实际要爬取的电商网站URLBASE_URL = "https://example-ecommerce-site.com/products"  # 替换为实际URLMAX_PAGES = 10  # 要爬取的页数NUM_THREADS = 5  # 线程数量# 创建并运行爬虫crawler = EcommerceCrawler(BASE_URL, MAX_PAGES, NUM_THREADS)crawler.run()

代码解析

上述代码实现了一个通用的多线程电商数据爬虫，主要包含以下几个核心部分：

初始化配置：设置基础 URL、最大爬取页数、线程数量等参数，初始化任务队列
请求处理：
- 随机生成请求头，模拟不同浏览器
- 发送 HTTP 请求获取页面内容
- 处理请求异常和错误状态码
页面解析：
- 使用 BeautifulSoup 解析 HTML
- 提取商品名称、价格、评分等关键信息
- 注意：实际使用时需要根据目标网站的 HTML 结构调整选择器
多线程实现：
- 使用 queue 创建任务队列，存放待爬取的页面
- 多个线程从队列中获取任务并执行
- 使用线程锁确保共享数据的安全访问
结果处理：
- 将爬取的商品信息保存到 CSV 文件
- 记录日志，便于监控和调试