京东商品价格爬虫实战：破解动态加载与反爬机制的完整指南
目录
前言：为什么选择爬取京东价格？
一、技术选型：为什么是这个组合？
1.1 动态加载的解决方案
1.2 完整技术栈
1.3 环境准备
二、破解京东反爬的十层防护
三、完整代码实现
3.1 浏览器配置类
3.2 价格提取器
3.3 批量爬取与代理管理
3.4 防封号策略增强版
3.5 主程序与使用示例
四、踩坑实录与解决方案
4.1 坑一：无头模式被检测
4.2 坑二：chromedriver版本不匹配
4.3 坑三：价格明明在页面上但就是抓不到
4.4 坑四：京东的随机class名
4.5 坑五：触发验证码后被无限重定向
4.6 坑六：Cookie过期后无法获取价格
五、性能优化与分布式扩展
5.1 单机优化
5.2 分布式爬虫架构
六、法律与道德提醒
前言：为什么选择爬取京东价格？
在电商数据分析、价格监控、竞品研究等领域，获取商品价格是最基础也最关键的一步。京东作为中国最大的B2C电商平台之一，其商品数据价值极高。然而，京东采用了典型的动态加载技术，传统的requests+BeautifulSoup方式会直接碰壁——你会发现页面源代码里根本找不到价格数字。
这篇文章将带你从零开始，用2025年最新的爬虫技术，完整实现一个能够稳定爬取京东商品价格的爬虫系统。我会把踩过的坑、遇到的验证码、被封IP的教训都分享出来，争取让你看完就能动手写一个能用的爬虫。
一、技术选型：为什么是这个组合？
1.1 动态加载的解决方案
京东商品页面是这样的：你打开一个商品页，看到一个价格数字，但查看网页源代码搜索这个数字——找不到。价格是通过JavaScript异步请求加载的。传统的requests.get()只能拿到HTML骨架。
解决方案有三种：
分析Ajax接口：最优雅，效率最高，但需要逆向JS逻辑
Selenium/Playwright模拟浏览器：最暴力，最稳定，适合初学者
Pyppeteer：异步版Puppeteer，介于两者之间
我选择Selenium作为主力。为什么？因为京东的反爬会检测webdriver特征，但我们可以通过参数规避。而Ajax接口京东经常换参数加密方式，维护成本太高。
1.2 完整技术栈
python
selenium==4.15.0 # 浏览器自动化 webdriver-manager==4.0.1 # 自动管理chromedriver pandas==2.1.0 # 数据存储 fake-useragent==1.4.0 # 随机UA retrying==1.3.4 # 重试机制 time, random, re # 标准库
1.3 环境准备
首先得安装Chrome浏览器。然后用pip安装依赖：
bash
pip install selenium webdriver-manager pandas fake-useragent retrying
webdriver-manager是个好东西，它会自动下载匹配你Chrome版本的驱动，不用手动去找了。
二、破解京东反爬的十层防护
在写代码之前，得先理解京东会怎么拦你。我在实际爬取中遇到的障碍包括：
检测webdriver属性- 京东的JS会检查
window.navigator.webdriver是不是true请求频率限制- 一秒超过2次就弹验证码
Cookie失效- 不携带有效cookie，返回登录页面
IP黑名单- 短时间内大量请求，IP直接被ban几个小时
User-Agent校验- 非浏览器UA直接拒绝
Referer检查- 空Referer或者异常Referer会被标记
x-requested-with头- Ajax请求缺少这个头
加密参数- 部分接口需要生成fingerprint
滑块验证码- 触发频率限制后弹出
账号风控- 用爬虫程序的cookie登录会被限制
针对这些，我们的应对策略会在代码中一一体现。
三、完整代码实现
3.1 浏览器配置类
这是最关键的部分。我们需要启动一个看起来完全像真实用户的Chrome实例。
python
# config.py import random from selenium import webdriver from selenium.webdriver.chrome.options import Options from selenium.webdriver.chrome.service import Service from webdriver_manager.chrome import ChromeDriverManager from fake_useragent import UserAgent class ChromeDriverConfig: """配置一个能骗过京东反爬的Chrome浏览器""" @staticmethod def get_driver(headless=False, proxy=None): """ 获取配置好的driver :param headless: 是否无头模式（建议False，因为无头模式更容易被检测） :param proxy: 代理地址，如 "http://127.0.0.1:8080" """ options = Options() # 1. 随机User-Agent，京东会根据UA做初步筛选 ua = UserAgent() random_ua = ua.random options.add_argument(f'user-agent={random_ua}') # 2. 隐藏webdriver特征 - 最重要的一步 # 这个实验性的参数会移除navigator.webdriver属性 options.add_experimental_option('excludeSwitches', ['enable-automation']) options.add_experimental_option('useAutomationExtension', False) # 3. 禁用自动化提示条 options.add_argument('--disable-blink-features=AutomationControlled') # 4. 设置窗口大小，避免小窗口被识别为爬虫 options.add_argument('--window-size=1920,1080') # 5. 禁用GPU加速，减少资源占用 options.add_argument('--disable-gpu') # 6. 禁用沙箱模式（Docker环境下需要） options.add_argument('--no-sandbox') # 7. 禁用共享内存（Linux环境下） options.add_argument('--disable-dev-shm-usage') # 8. 屏蔽通知弹窗 options.add_argument('--disable-notifications') # 9. 禁用插件 options.add_argument('--disable-plugins') # 10. 设置语言 options.add_argument('--lang=zh-CN') # 11. 设置默认编码 options.add_argument('--accept-encoding=utf-8') # 12. 忽略证书错误 options.add_argument('--ignore-certificate-errors') # 13. 允许所有Cookie options.add_argument('--enable-features=NetworkService,NetworkServiceInProcess') # 14. 无头模式（可选，但容易被检测） if headless: options.add_argument('--headless=new') # 无头模式下需要额外设置，避免被检测 options.add_argument('--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36') # 15. 代理配置 if proxy: options.add_argument(f'--proxy-server={proxy}') # 16. 常用预制选项，使浏览器更像真实用户 preferences = { 'profile.default_content_setting_values.notifications': 2, # 禁用通知 'credentials_enable_service': False, # 禁用保存密码 'profile.password_manager_enabled': False, # 禁用密码管理器 'profile.default_content_settings.popups': 0, # 禁止弹窗 'download.prompt_for_download': False, # 自动下载 'download.default_directory': '/dev/null', # 下载目录 } options.add_experimental_option('prefs', preferences) # 创建service，使用webdriver-manager自动管理驱动 service = Service(ChromeDriverManager().install()) # 创建driver driver = webdriver.Chrome(service=service, options=options) # 执行脚本来隐藏webdriver痕迹（另一种方式） driver.execute_cdp_cmd('Page.addScriptToEvaluateOnNewDocument', { 'source': ''' Object.defineProperty(navigator, 'webdriver', { get: () => undefined }); Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3, 4, 5] }); Object.defineProperty(navigator, 'languages', { get: () => ['zh-CN', 'zh', 'en'] }); window.chrome = { runtime: {} }; ''' }) return driver3.2 价格提取器
价格在页面里藏得很深。京东的价格有两种呈现方式：
普通价格：直接在某个
span或div里促销价格：有时需要触发价格组件才能加载
python
# price_extractor.py import re import time from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from selenium.common.exceptions import TimeoutException, NoSuchElementException class JDPriceExtractor: """京东商品价格提取器""" def __init__(self, driver, timeout=10): self.driver = driver self.timeout = timeout self.wait = WebDriverWait(driver, timeout) def extract_price_from_detail_page(self, sku_id): """ 从商品详情页提取价格 :param sku_id: 京东商品ID :return: dict包含价格信息 """ url = f'https://item.jd.com/{sku_id}.html' self.driver.get(url) # 随机延迟，模拟人类行为 time.sleep(random.uniform(2, 4)) result = { 'sku_id': sku_id, 'url': url, 'price': None, 'original_price': None, 'promotion_info': None, 'success': False } try: # 方法1：等待价格元素出现 - 最常见的价格位置 price_selectors = [ '#price > div > div > span.price', # 新版页面 '#jd-price', # 旧版页面 '.price.J-p-{}'.format(sku_id), # 动态类名 'span.p-price', # 促销价格 'div.summary-price > div > div > span', # 总结区域价格 'div.sku-price > span' # SKU价格 ] price_element = None for selector in price_selectors: try: price_element = self.wait.until( EC.presence_of_element_located((By.CSS_SELECTOR, selector)) ) if price_element and price_element.text.strip(): break except: continue if price_element: price_text = price_element.text.strip() # 提取数字，处理像"¥199.00"或"199.00"这样的格式 price_match = re.search(r'[\d,]+\.?\d*', price_text) if price_match: result['price'] = float(price_match.group().replace(',', '')) # 方法2：获取原价（划线价） original_price_selectors = [ 'del.J-p-{}'.format(sku_id), 'span.price.J-price', 'div.summary-price > div > del', 's.price' ] for selector in original_price_selectors: try: orig_element = self.driver.find_element(By.CSS_SELECTOR, selector) if orig_element and orig_element.text.strip(): price_text = orig_element.text.strip() price_match = re.search(r'[\d,]+\.?\d*', price_text) if price_match: result['original_price'] = float(price_match.group().replace(',', '')) break except: continue # 方法3：检测促销信息（满减、优惠券等） try: promotion_selectors = [ 'div.promo-tag', 'div.promo-tag-list', 'div.discount-wrap', 'div.j-quantity-fake-discount' ] promo_elements = [] for selector in promotion_selectors: promo_elements.extend(self.driver.find_elements(By.CSS_SELECTOR, selector)) if promo_elements: result['promotion_info'] = [elem.text.strip() for elem in promo_elements if elem.text.strip()] except: pass # 判断是否成功 if result['price']: result['success'] = True else: # 可能遇到验证码页面 if self._check_captcha_page(): result['success'] = False result['error'] = 'Captcha detected' else: result['success'] = False result['error'] = 'Price not found' except TimeoutException: result['error'] = 'Timeout waiting for price element' except Exception as e: result['error'] = str(e) return result def extract_price_from_search_page(self, sku_id): """ 另一种方法：通过搜索页面获取价格 有时候详情页反爬太强，但搜索页还能抓 """ search_url = f'https://search.jd.com/Search?keyword={sku_id}' self.driver.get(search_url) time.sleep(random.uniform(2, 3)) try: # 搜索页的价格通常在li元素里 product_items = self.wait.until( EC.presence_of_all_elements_located((By.CSS_SELECTOR, 'li.gl-item')) ) for item in product_items: # 检查SKU是否匹配 try: sku_elem = item.find_element(By.CSS_SELECTOR, 'div[data-sku]') item_sku = sku_elem.get_attribute('data-sku') if item_sku == sku_id: price_elem = item.find_element(By.CSS_SELECTOR, 'div.p-price i.price') price_text = price_elem.text.strip() price_match = re.search(r'[\d,]+\.?\d*', price_text) if price_match: return float(price_match.group().replace(',', '')) except: continue except: pass return None def _check_captcha_page(self): """检查是否遇到了验证码页面""" page_source = self.driver.page_source captcha_keywords = ['captcha', '验证码', '滑块', 'verify'] for keyword in captcha_keywords: if keyword in page_source.lower(): return True return False def get_price_via_ajax(self, sku_id): """ 终极方法：直接请求价格接口 这个方法需要分析京东的前端接口，这里给出思路 """ # 京东的价格接口示例（注意：这个接口随时可能变） # 通常格式：https://p.3.cn/prices/mgets?skuIds=J_{sku_id} import requests api_url = f'https://p.3.cn/prices/mgets?skuIds=J_{sku_id}' headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36', 'Referer': f'https://item.jd.com/{sku_id}.html', 'Accept': 'application/json, text/plain, */*' } try: response = requests.get(api_url, headers=headers, timeout=5) if response.status_code == 200: data = response.json() if data and 'p' in data[0]: return float(data[0]['p']) except: pass return None3.3 批量爬取与代理管理
实际应用中，我们往往需要爬取成百上千个商品。这时候单线程+单IP必死无疑。
python
# batch_crawler.py import random import time import json import pandas as pd from datetime import datetime from concurrent.futures import ThreadPoolExecutor, as_completed from retrying import retry from selenium.webdriver.common.by import By class JDBatchCrawler: """京东批量价格爬虫""" def __init__(self, driver_config, max_workers=3, request_interval=3): """ :param driver_config: ChromeDriverConfig类 :param max_workers: 并发数（不要太高，京东会封） :param request_interval: 请求间隔秒数 """ self.driver_config = driver_config self.max_workers = max_workers self.request_interval = request_interval self.results = [] self.failed_skus = [] @retry(stop_max_attempt_number=3, wait_fixed=2000) def crawl_single_sku(self, sku_id, proxy=None): """ 爬取单个商品，带重试机制 """ driver = None try: driver = self.driver_config.get_driver(proxy=proxy) extractor = JDPriceExtractor(driver) result = extractor.extract_price_from_detail_page(sku_id) # 添加时间戳 result['crawl_time'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S') # 随机延时，避免频率过高 time.sleep(random.uniform(1, self.request_interval)) return result except Exception as e: print(f'Error crawling {sku_id}: {str(e)}') return { 'sku_id': sku_id, 'success': False, 'error': str(e), 'crawl_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S') } finally: if driver: driver.quit() def crawl_batch(self, sku_list, proxy_list=None, callback=None): """ 批量爬取 :param sku_list: 商品ID列表 :param proxy_list: 代理列表，轮换使用 :param callback: 进度回调函数 """ total = len(sku_list) completed = 0 with ThreadPoolExecutor(max_workers=self.max_workers) as executor: futures = {} for idx, sku_id in enumerate(sku_list): # 轮换代理 proxy = None if proxy_list: proxy = proxy_list[idx % len(proxy_list)] future = executor.submit(self.crawl_single_sku, sku_id, proxy) futures[future] = sku_id for future in as_completed(futures): result = future.result() self.results.append(result) if not result.get('success'): self.failed_skus.append(result['sku_id']) completed += 1 if callback: callback(completed, total) # 在大批量爬取时，每50个休息10秒 if completed % 50 == 0: print(f'已完成 {completed}/{total}，休息10秒...') time.sleep(10) return self.results def save_to_excel(self, filename='jd_prices.xlsx'): """保存结果到Excel""" df = pd.DataFrame(self.results) df.to_excel(filename, index=False) print(f'结果已保存到 {filename}') return df def save_to_json(self, filename='jd_prices.json'): """保存结果到JSON""" with open(filename, 'w', encoding='utf-8') as f: json.dump(self.results, f, ensure_ascii=False, indent=2) print(f'结果已保存到 {filename}') def generate_report(self): """生成爬取报告""" total = len(self.results) success = sum(1 for r in self.results if r.get('success')) failed = total - success report = f""" ========== 京东价格爬取报告 ========== 爬取时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} 总商品数: {total} 成功数量: {success} ({success/total*100:.1f}%) 失败数量: {failed} ({failed/total*100:.1f}%) 并发线程: {self.max_workers} 请求间隔: {self.request_interval}秒 价格统计（成功的数据）: """ if success > 0: prices = [r['price'] for r in self.results if r.get('price')] if prices: report += f""" 最高价格: ¥{max(prices):.2f} 最低价格: ¥{min(prices):.2f} 平均价格: ¥{sum(prices)/len(prices):.2f} """ report += f""" 失败商品ID列表（前20个）: {self.failed_skus[:20]} """ print(report) return report3.4 防封号策略增强版
上面的代码已经实现基础功能，但要想长时间稳定运行，还需要一个专门的防封模块。
python
# anti_ban.py import random import time import pickle import os from selenium.webdriver.common.by import By class AntiBanManager: """反反爬虫管理器""" def __init__(self, driver): self.driver = driver self.cookie_file = 'jd_cookies.pkl' def human_like_behavior(self): """ 模拟人类行为： - 随机滚动页面 - 随机鼠标移动 - 随机等待时间 """ # 随机滚动到页面不同位置 scroll_positions = [100, 300, 500, 800, 1200, 'bottom'] scroll_to = random.choice(scroll_positions) if scroll_to == 'bottom': self.driver.execute_script("window.scrollTo(0, document.body.scrollHeight);") else: self.driver.execute_script(f"window.scrollTo(0, {scroll_to});") # 随机停留 time.sleep(random.uniform(0.5, 2)) # 10%的概率会鼠标移动（模拟） if random.random() < 0.1: # 这里可以用ActionChains实现更真实的鼠标移动 pass # 5%的概率会回滚到顶部 if random.random() < 0.05: self.driver.execute_script("window.scrollTo(0, 0);") time.sleep(random.uniform(0.3, 1)) def load_cookies(self): """加载之前保存的cookies""" if os.path.exists(self.cookie_file): try: with open(self.cookie_file, 'rb') as f: cookies = pickle.load(f) for cookie in cookies: # 处理过期cookie if 'expiry' in cookie: cookie['expiry'] = int(cookie['expiry']) self.driver.add_cookie(cookie) print(f'成功加载 {len(cookies)} 个cookies') return True except Exception as e: print(f'加载cookies失败: {e}') return False def save_cookies(self): """保存cookies供下次使用""" try: cookies = self.driver.get_cookies() with open(self.cookie_file, 'wb') as f: pickle.dump(cookies, f) print(f'成功保存 {len(cookies)} 个cookies') return True except Exception as e: print(f'保存cookies失败: {e}') return False def login_if_needed(self, username=None, password=None): """ 如果需要登录，执行登录操作 注意：京东的登录有滑块验证，手动登录更靠谱 """ # 检查是否需要登录 if 'passport' in self.driver.current_url or 'login' in self.driver.current_url: print('检测到需要登录，请手动扫码登录...') # 等待用户手动登录（给你60秒时间扫码） for i in range(60): time.sleep(1) if 'passport' not in self.driver.current_url and 'login' not in self.driver.current_url: print('登录成功！') self.save_cookies() return True print('登录超时') return False return True def random_delay(self, min_sec=1, max_sec=3): """随机延迟，避免规律性请求""" delay = random.uniform(min_sec, max_sec) time.sleep(delay) def check_ip_blocked(self): """检查IP是否被封""" page_source = self.driver.page_source.lower() blocked_keywords = ['拒绝访问', 'access denied', '验证', 'captcha', '滑块'] for keyword in blocked_keywords: if keyword in page_source: print(f'检测到可能被封: {keyword}') return True return False3.5 主程序与使用示例
python
# main.py import random import sys from colorama import init, Fore, Style # 可选，用于彩色输出 # 初始化colorama init(autoreset=True) def print_banner(): """打印程序横幅""" banner = f""" {Fore.CYAN}╔══════════════════════════════════════════════════════════╗ ║ 京东商品价格爬虫 v2.0 - 动态加载破解版 ║ ║ 支持批量爬取 | 自动换IP | 反反爬虫策略 ║ ╚══════════════════════════════════════════════════════════╝{Style.RESET_ALL} """ print(banner) def show_progress(current, total): """显示进度条""" percent = current / total * 100 bar_length = 40 filled = int(bar_length * current // total) bar = '█' * filled + '░' * (bar_length - filled) sys.stdout.write(f'\r进度: |{bar}| {percent:.1f}% ({current}/{total})') sys.stdout.flush() def main(): print_banner() # ========== 配置区域 ========== # 要爬取的商品ID列表（可以从Excel或数据库读取） sku_list = [ '100012043978', # 示例：iPhone 15 '100038009238', # 示例：某款笔记本 '100012345678', # 替换成你需要的商品ID # 可以添加更多... ] # 代理列表（可选，建议购买付费代理） proxy_list = [ # 'http://user:pass@ip:port', # 'http://user:pass@ip:port', ] # 爬虫参数 MAX_WORKERS = 2 # 并发数，京东很敏感，建议从1或2开始 REQUEST_INTERVAL = 5 # 请求间隔秒数，越大越安全 # ========== 开始爬取 ========== print(f'{Fore.YELLOW}[INFO] 开始爬取，共 {len(sku_list)} 个商品') print(f'[INFO] 并发数: {MAX_WORKERS}, 请求间隔: {REQUEST_INTERVAL}秒{Style.RESET_ALL}') # 创建爬虫实例 crawler = JDBatchCrawler( driver_config=ChromeDriverConfig, max_workers=MAX_WORKERS, request_interval=REQUEST_INTERVAL ) # 执行批量爬取 results = crawler.crawl_batch( sku_list=sku_list, proxy_list=proxy_list if proxy_list else None, callback=show_progress ) print('\n') # 换行 # ========== 保存结果 ========== if results: # 保存为Excel df = crawler.save_to_excel('jd_prices_{}.xlsx'.format( datetime.now().strftime('%Y%m%d_%H%M%S') )) # 保存为JSON crawler.save_to_json('jd_prices_{}.json'.format( datetime.now().strftime('%Y%m%d_%H%M%S') )) # 生成报告 crawler.generate_report() # 打印成功的结果 print(f'\n{Fore.GREEN}[SUCCESS] 爬取完成！成功爬取到价格的商品:{Style.RESET_ALL}') success_results = [r for r in results if r.get('success')] for r in success_results[:10]: # 只显示前10个 print(f' 商品 {r["sku_id"]}: ¥{r["price"]} (原价: {r.get("original_price", "无")})') if len(success_results) > 10: print(f' ... 共{len(success_results)}个成功') else: print(f'{Fore.RED}[ERROR] 没有获取到任何数据{Style.RESET_ALL}') def test_single_product(): """测试单个商品，用于调试""" sku_id = input('请输入商品ID: ').strip() print(f'正在爬取商品 {sku_id} ...') driver = ChromeDriverConfig.get_driver() extractor = JDPriceExtractor(driver) result = extractor.extract_price_from_detail_page(sku_id) if result['success']: print(f'✅ 成功！价格: ¥{result["price"]}') if result.get('original_price'): print(f'原价: ¥{result["original_price"]}') if result.get('promotion_info'): print(f'促销信息: {result["promotion_info"]}') else: print(f'❌ 失败: {result.get("error", "未知错误")}') driver.quit() if __name__ == '__main__': import argparse from datetime import datetime parser = argparse.ArgumentParser(description='京东商品价格爬虫') parser.add_argument('--mode', choices=['batch', 'test'], default='batch', help='运行模式: batch批量爬取, test测试单个商品') args = parser.parse_args() if args.mode == 'test': test_single_product() else: main()四、踩坑实录与解决方案
写这个爬虫的过程中，我遇到了无数坑，这里把最典型的几个列出来。
4.1 坑一：无头模式被检测
最开始我用headless=True，心想这样节省资源。结果京东直接返回空白页面或者验证码。
原因：无头模式下navigator.webdriver属性为true，京东的JS能检测到。
解决方案：
放弃无头模式，用有头模式
如果一定要无头，需要添加
--headless=new并配合上面代码中的CDP脚本
4.2 坑二：chromedriver版本不匹配
有一次Chrome自动更新了，我的爬虫突然报错session not created。折腾了半天发现是驱动版本不对。
解决方案：用webdriver-manager自动管理，再也不用担心版本问题。
4.3 坑三：价格明明在页面上但就是抓不到
观察发现，商品价格不是一开始就加载的，需要滚动到价格区域才会触发加载。
解决方案：在等待价格元素之前，先执行一个滚动动作到价格区域附近。
python
# 滚动到价格可能出现的位置 self.driver.execute_script("window.scrollTo(0, 300);") time.sleep(0.5)4.4 坑四：京东的随机class名
京东给价格元素加的是动态class，比如p-price-9f3k2d1，每次刷新都不一样。
解决方案：用属性选择器或者部分匹配，比如[class*="p-price"]，或者用稳定的ID如#jd-price。
4.5 坑五：触发验证码后被无限重定向
一旦触发验证码，京东会让你一直验证，即使你关了浏览器重新开，IP已经被标记了。
解决方案：
使用高质量的代理IP
降低请求频率
实现验证码识别（这个成本较高，建议直接换IP）
4.6 坑六：Cookie过期后无法获取价格
Cookie默认有效期大概24小时，过期后访问商品页会跳转到登录页。
解决方案：
实现Cookie持久化存储和加载
定时登录更新Cookie（但京东登录有滑块，建议手动定期更新）
五、性能优化与分布式扩展
5.1 单机优化
复用driver：不要每个商品都启动关闭浏览器，一个driver可以爬多个商品
减少等待时间：将显式等待的超时时间从10秒降到5秒
使用CSS选择器而不是XPath：CSS选择器效率更高
5.2 分布式爬虫架构
当商品数量达到万级以上，单机不够用了。可以考虑：
Redis作为URL队列：Master节点分发任务，Worker节点消费
代理池：用Redis维护一个代理IP池，自动检测可用性
数据库存储：MySQL/PostgreSQL存储结果，避免重复爬取
下面是一个简化的分布式任务队列实现：
python
# distributed_queue.py import redis import json class RedisTaskQueue: """基于Redis的分布式任务队列""" def __init__(self, host='localhost', port=6379, db=0): self.redis_client = redis.Redis(host=host, port=port, db=db) self.task_key = 'jd:price:tasks' self.result_key = 'jd:price:results' def push_tasks(self, sku_list): """批量推送任务""" for sku_id in sku_list: self.redis_client.lpush(self.task_key, sku_id) print(f'已推送 {len(sku_list)} 个任务') def pop_task(self, timeout=5): """获取任务（阻塞模式）""" result = self.redis_client.brpop(self.task_key, timeout=timeout) if result: return result[1].decode('utf-8') return None def save_result(self, sku_id, price_data): """保存爬取结果""" data = { 'sku_id': sku_id, 'price_data': price_data, 'timestamp': datetime.now().isoformat() } self.redis_client.lpush(self.result_key, json.dumps(data)) def get_results(self): """获取所有结果""" results = [] while True: result = self.redis_client.rpop(self.result_key) if not result: break results.append(json.loads(result)) return results六、法律与道德提醒
写爬虫之前，有几点必须要说明：
遵守robots.txt：京东的
https://www.jd.com/robots.txt明确禁止了部分爬取行为。严格来说，批量爬取商品数据可能违反服务条款。控制请求频率：不要对京东服务器造成压力，这是基本礼貌。
数据使用：爬下来的价格数据不要用于商业目的，尤其是不要做比价平台与京东直接竞争。
IP代理：使用代理时，确保代理来源合法。
尊重用户隐私：不要爬取用户评论中的个人信息。