避免爬虫无限循环：分页链接识别与处理实战指南

分页是爬虫采集列表数据（商品、新闻、评论）的核心场景，但“重复分页”“动态分页”“虚假分页”等问题极易导致爬虫陷入无限循环（如反复爬取同一页）或遗漏数据。本文从“分页类型识别→去重策略→终止条件→实战案例”四个维度，拆解分页链接的高效处理方案，帮你彻底解决无限循环问题。

一、先识别：常见分页类型与核心特征

不同网站的分页实现差异极大，先明确分页类型，再针对性处理，是避免循环的前提：

分页类型	核心特征	链接格式示例	识别要点
静态页码分页（最常见）	URL含页码参数（page/no），页码递增，有明确总页数	?page=1 &no=2 page/3	URL含 page/ no/ pageNum等关键词；页面显示“共10页”
偏移量分页	URL含偏移量（offset）+ 每页条数（limit）	?offset=0&limit=20 ?start=20&count=20	偏移量按 limit递增（如0→20→40）；无明确页码
动态加载分页（AJAX）	无URL变化，滚动/点击“加载更多”触发AJAX请求	接口返回 {"has_more": true, "data": [...]}	页面无页码，只有“加载更多”按钮；Network中XHR请求含分页参数
滚动加载分页（无限滚动）	滚动到页面底部自动加载下一页，无页码/按钮	接口参数 ?last_id=xxx ?cursor=xxx	依赖上一页最后一条数据的ID/cursor作为下一页参数
虚假分页（陷阱）	页码递增但内容重复，或URL变化但数据不变	?page=1与 ?page=2返回相同数据	多页数据MD5值相同；页码超过实际页数仍返回数据

二、核心原则：避免无限循环的3个关键

无论哪种分页类型，处理的核心都是“去重+终止”，再结合分页特征优化，三者缺一不可：

链接去重：确保同一分页链接只爬取一次（核心防循环）；明确终止条件：爬取到“最后一页”时立即停止（核心防无限爬取）；适配分页特征：按分页类型构造下一页链接（核心防遗漏）。

三、实战方案：分类型处理分页链接

3.1 静态页码分页（URL含page参数）

最易处理的分页类型，核心是“提取页码→去重→判断总页数”。

核心步骤：

提取分页参数：从URL中解析页码（如 page=1），明确每页条数；去重存储：用集合/Redis记录已爬取的页码/链接，避免重复；终止条件： 显式终止：页码超过页面显示的“总页数”（如“共10页”，爬完page=10停止）；隐式终止：下一页链接不存在（如page=11返回404）；当前页数据为空； 构造下一页链接：页码+1，替换URL中的分页参数。

代码实现：

import requests
from bs4 import BeautifulSoup
from urllib.parse import urlparse, parse_qs, urlencode, urlunparse
import hashlib

# 1. 初始化去重集合（小规模用集合，大规模用Redis）
crawled_pages = set()
# 目标列表页URL（含分页参数）
base_url = "https://example.com/products"
page_param = "page"  # 分页参数名（根据实际URL调整）
per_page = 20  # 每页条数（可从页面提取或硬编码）

def get_total_pages(soup):
    """从页面提取总页数（示例：页面显示“共10页”）"""
    try:
        # 按实际页面结构调整选择器（常见：页码容器、总页数文本）
        total_text = soup.select_one("div.pagination span.total").get_text(strip=True)
        # 正则提取数字（如“共10页”→10）
        total_pages = int("".join(filter(str.isdigit, total_text)))
        return total_pages
    except Exception as e:
        print(f"提取总页数失败，将通过隐式条件终止：{e}")
        return None

def parse_page_url(url, page_num):
    """构造指定页码的URL"""
    # 解析URL components
    parsed = urlparse(url)
    # 解析查询参数
    query_params = parse_qs(parsed.query)
    # 更新页码参数
    query_params[page_param] = [str(page_num)]
    # 重新构造查询字符串
    new_query = urlencode(query_params, doseq=True)
    # 重新构造URL
    new_parsed = parsed._replace(query=new_query)
    return urlunparse(new_parsed)

def is_duplicate_page(html):
    """判断当前页是否为重复数据（防虚假分页）"""
    # 计算HTML核心数据区域的MD5值（避免因广告等无关内容误判）
    soup = BeautifulSoup(html, "lxml")
    core_data = soup.select_one("div.product-list").get_text(strip=True) if soup.select_one("div.product-list") else ""
    data_md5 = hashlib.md5(core_data.encode("utf-8")).hexdigest()
    # 若MD5已存在，说明是重复页
    if data_md5 in crawled_pages:
        return True
    crawled_pages.add(data_md5)
    return False

def crawl_static_pagination():
    current_page = 1
    total_pages = None
    
    while True:
        # 1. 构造当前页URL
        current_url = parse_page_url(base_url, current_page)
        print(f"正在爬取：{current_url}")
        
        # 2. 跳过已爬取的URL（双重去重：URL+数据MD5）
        if current_url in crawled_pages:
            print(f"URL已爬取，跳过：{current_url}")
            current_page += 1
            continue
        
        # 3. 发送请求
        headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}
        try:
            response = requests.get(current_url, headers=headers, timeout=10)
            response.raise_for_status()  # 抛出404/500等错误
        except Exception as e:
            print(f"爬取失败：{e}")
            break
        
        # 4. 检查是否为重复数据（防虚假分页）
        if is_duplicate_page(response.text):
            print(f"重复数据页，终止爬取")
            break
        
        # 5. 解析页面数据（商品/新闻等）
        soup = BeautifulSoup(response.text, "lxml")
        parse_data(soup)  # 自定义数据解析函数（略）
        
        # 6. 记录已爬取URL
        crawled_pages.add(current_url)
        
        # 7. 提取总页数（仅第一次爬取时提取）
        if not total_pages:
            total_pages = get_total_pages(soup)
        
        # 8. 判断是否终止爬取
        if total_pages:
            # 显式终止：当前页≥总页数
            if current_page >= total_pages:
                print(f"已爬取到最后一页（共{total_pages}页），终止")
                break
        else:
            # 隐式终止：检查是否有下一页按钮/链接
            next_page_btn = soup.select_one("a.next-page")  # 下一页按钮选择器
            if not next_page_btn or "disabled" in next_page_btn.get("class", []):
                print(f"无下一页，终止爬取")
                break
        
        # 9. 进入下一页
        current_page += 1

def parse_data(soup):
    """解析页面数据（示例：提取商品名称）"""
    products = soup.select("div.product-item")
    print(f"当前页提取{len(products)}条数据")
    # 实际数据提取逻辑（略）

# 启动爬取
crawl_static_pagination()

关键优化：

双重去重：URL去重（避免重复请求）+ 数据MD5去重（避免虚假分页）；容错处理：若无法提取总页数，用“下一页按钮是否存在”作为备用终止条件；参数兼容：处理URL中分页参数的不同位置（如 ?page=1&sort=price），用 urllib.parse解析重构，避免手动拼接出错。

3.2 偏移量分页（offset+limit）

无明确页码，核心是“偏移量递增→判断数据是否为空”。

核心步骤：

提取参数：从URL/接口中解析 offset（偏移量）和 limit（每页条数）；构造下一页：下一页偏移量 = 当前偏移量 + limit；终止条件：当前页数据为空（无新数据）；接口返回 has_more=false；去重：记录已爬取的偏移量，或通过数据唯一ID（如商品ID）去重。

代码实现（以API接口为例）：

import requests
import json

crawled_offsets = set()
api_url = "https://example.com/api/products"
limit = 20  # 每页条数（从接口文档/响应中获取）

def crawl_offset_pagination():
    offset = 0  # 初始偏移量
    
    while True:
        # 1. 构造请求参数
        params = {
            "offset": offset,
            "limit": limit,
            "sort": "latest"
        }
        
        # 2. 去重：跳过已爬取的偏移量
        if offset in crawled_offsets:
            print(f"偏移量{offset}已爬取，跳过")
            offset += limit
            continue
        
        # 3. 发送请求
        headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}
        try:
            response = requests.get(api_url, params=params, headers=headers, timeout=10)
            data = response.json()
        except Exception as e:
            print(f"请求失败：{e}")
            break
        
        # 4. 解析数据与终止判断
        products = data.get("data", [])
        if not products:
            print(f"当前偏移量{offset}无数据，终止爬取")
            break
        
        # 5. 数据去重（通过商品ID避免重复）
        product_ids = [p["id"] for p in products]
        if len(set(product_ids)) != len(product_ids):
            print(f"当前页存在重复数据，终止爬取")
            break
        
        # 6. 处理数据
        print(f"偏移量{offset}提取{len(products)}条数据")
        # 数据存储逻辑（略）
        
        # 7. 记录已爬取偏移量
        crawled_offsets.add(offset)
        
        # 8. 检查是否有更多数据（接口返回标识）
        has_more = data.get("has_more", False)
        if not has_more:
            print(f"接口返回无更多数据，终止爬取")
            break
        
        # 9. 进入下一页（偏移量递增）
        offset += limit

# 启动爬取
crawl_offset_pagination()

3.3 动态加载分页（加载更多按钮/AJAX）

URL无变化，分页逻辑隐藏在AJAX接口中，核心是“模拟触发→解析接口参数→判断 has_more”。

核心步骤：

找到AJAX接口：用Chrome开发者工具（Network→XHR）捕获“加载更多”触发的请求；提取接口参数：记录分页参数（如 page/ last_id）和请求头（如 Referer/ Cookie）；模拟请求：用Requests直接调用AJAX接口（比Selenium高效）；终止条件：接口返回 has_more=false；“加载更多”按钮不可点击；去重：记录已爬取的分页参数（如 page）或数据ID。

代码实现（结合Selenium获取Cookie）：

import requests
import json
from selenium import webdriver
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.by import By

# 1. 用Selenium获取登录后的Cookie（如需登录）
def get_cookies():
    driver = webdriver.Chrome()
    driver.get("https://example.com/login")
    # 模拟登录（略）
    cookies = driver.get_cookies()
    driver.quit()
    return {c["name"]: c["value"] for c in cookies}

# 2. 爬取AJAX动态分页
def crawl_ajax_pagination():
    cookies = get_cookies()
    ajax_api = "https://example.com/api/load-more"
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
        "Referer": "https://example.com/products",
        "Content-Type": "application/json"
    }
    
    page = 1
    crawled_pages = set()
    
    while True:
        # 1. 构造AJAX请求参数
        data = {
            "page": page,
            "pageSize": 20,
            "sortType": 1
        }
        
        # 2. 去重
        if page in crawled_pages:
            print(f"第{page}页已爬取，跳过")
            page += 1
            continue
        
        # 3. 发送POST请求（AJAX接口多为POST）
        try:
            response = requests.post(
                ajax_api,
                headers=headers,
                cookies=cookies,
                json=data,
                timeout=10
            )
            result = response.json()
        except Exception as e:
            print(f"请求失败：{e}")
            break
        
        # 4. 解析数据与终止判断
        if result.get("code") != 200:
            print(f"接口返回错误：{result.get('msg')}")
            break
        
        products = result.get("data", {}).get("list", [])
        if not products:
            print(f"第{page}页无数据，终止爬取")
            break
        
        # 5. 处理数据
        print(f"第{page}页提取{len(products)}条数据")
        # 数据存储逻辑（略）
        
        # 6. 记录已爬取页码
        crawled_pages.add(page)
        
        # 7. 检查是否有更多数据
        has_more = result.get("data", {}).get("hasMore", False)
        if not has_more:
            print(f"无更多数据，终止爬取")
            break
        
        # 8. 进入下一页
        page += 1

# 启动爬取
crawl_ajax_pagination()

3.4 滚动加载分页（无限滚动，cursor分页）

依赖上一页的“游标”（如最后一条数据的ID/时间戳），核心是“提取cursor→传递给下一页→判断cursor是否为空”。

核心步骤：

提取cursor：从当前页最后一条数据中获取 last_id/ cursor（如商品ID、评论时间戳）；构造下一页请求：将cursor作为参数传入下一页接口；终止条件：cursor为空；接口返回 has_more=false；下一页数据与当前页重复；去重：记录已爬取的cursor或数据唯一ID，避免循环。

代码实现：

import requests
import json

crawled_cursors = set()
api_url = "https://example.com/api/scroll-products"
headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}

def crawl_scroll_pagination():
    cursor = ""  # 初始游标（空表示第一页）
    
    while True:
        # 1. 构造请求参数（cursor为核心分页参数）
        params = {
            "cursor": cursor,
            "limit": 20
        }
        
        # 2. 去重：跳过已爬取的游标
        if cursor in crawled_cursors:
            print(f"游标{cursor}已爬取，避免循环，终止")
            break
        
        # 3. 发送请求
        try:
            response = requests.get(api_url, params=params, headers=headers, timeout=10)
            data = response.json()
        except Exception as e:
            print(f"请求失败：{e}")
            break
        
        # 4. 解析数据与终止判断
        products = data.get("items", [])
        if not products:
            print(f"无数据，终止爬取")
            break
        
        # 5. 数据去重（通过商品ID）
        product_ids = [p["id"] for p in products]
        if len(set(product_ids)) != len(product_ids):
            print(f"当前页存在重复数据，终止")
            break
        
        # 6. 处理数据
        print(f"游标{cursor}提取{len(products)}条数据")
        # 数据存储逻辑（略）
        
        # 7. 记录已爬取游标
        crawled_cursors.add(cursor)
        
        # 8. 提取下一页游标（从响应中获取）
        cursor = data.get("next_cursor", "")
        # 终止条件：无下一页游标
        if not cursor:
            print(f"无下一页游标，终止爬取")
            break

# 启动爬取
crawl_scroll_pagination()

四、关键技术：分页去重与终止条件的进阶优化

4.1 去重策略：从简单到复杂（按需选择）

去重方式	实现成本	适用场景	优点	缺点
集合存储已爬URL/页码	低	小规模爬取、静态分页	实现简单、查询快	内存占用随爬取页数增长
Redis存储（分布式去重）	中	大规模、分布式爬虫	支持多节点共享、内存高效	需部署Redis，增加依赖
数据唯一ID去重	中	所有分页类型（尤其是动态分页）	彻底避免重复数据，不受分页参数影响	需提取数据唯一ID（如商品ID、新闻ID）
数据MD5去重	中	虚假分页、数据重复场景	无需依赖分页参数，直接判断数据重复	需提取核心数据区域，计算MD5有开销

推荐组合：

小规模爬取：集合（URL/页码）+ 数据ID去重；大规模/分布式爬取：Redis（存储URL/游标）+ 数据ID去重；虚假分页场景：额外增加数据MD5去重。

4.2 终止条件：多重校验，避免误终止/无限循环

单一终止条件易出错（如接口 has_more返回异常），建议设置“多重终止校验”：

def check_terminate_conditions(data, products, current_page, crawled_count):
    """多重终止条件校验"""
    # 条件1：数据为空
    if not products:
        return True, "数据为空"
    # 条件2：已爬取页数超过阈值（防异常情况）
    if current_page > 100:  # 假设最多100页
        return True, "超过最大爬取页数阈值"
    # 条件3：接口返回无更多数据
    if not data.get("has_more", True):
        return True, "接口返回无更多数据"
    # 条件4：爬取数据量超过预期（防无限数据）
    if crawled_count > 10000:  # 假设最多爬10000条
        return True, "超过最大数据量阈值"
    # 条件5：数据重复率超过50%
    unique_ids = set([p["id"] for p in products])
    if len(unique_ids) / len(products) < 0.5:
        return True, "数据重复率过高"
    return False, ""

# 使用示例
crawled_count = 0
while True:
    # 爬取数据...
    products = data.get("items", [])
    crawled_count += len(products)
    # 校验终止条件
    should_terminate, reason = check_terminate_conditions(data, products, current_page, crawled_count)
    if should_terminate:
        print(f"终止爬取：{reason}")
        break

五、避坑指南：10个分页处理高频问题

URL参数顺序变化导致重复爬取→ 用 urllib.parse解析重构URL，确保参数顺序一致；**分页参数被URL编码（如page%3D1）**→ 解码后再去重（ urllib.parse.unquote(url)）；动态分页接口需要登录Cookie→ 用Selenium模拟登录获取Cookie，或用Requests.Session保持会话；**页码超过实际页数仍返回200（虚假分页）**→ 用数据MD5去重，或对比连续几页数据是否重复；滚动分页cursor重复导致循环→ 记录已爬取的cursor，遇到重复立即终止；“加载更多”按钮存在但点击无新数据→ 检查接口返回的 has_more字段，而非仅判断按钮是否存在；**分页参数是加密的（如page=xxx_sign）**→ 破解加密逻辑（参考“动态网页抓取”相关内容），或用Selenium模拟点击；多线程爬取导致重复分页→ 用Redis分布式锁，确保同一分页仅被一个线程爬取；爬取速度过快导致分页参数失效→ 增加随机延迟（ time.sleep(random.uniform(1, 3))），模拟人类操作；**页面分页与接口分页不一致（如页面显示10页，接口有20页）**→ 以接口返回的 has_more或数据为空作为最终终止条件。

六、实战案例：多类型分页混合处理（通用爬虫）

需求：

开发一个通用分页爬虫，支持静态页码、偏移量、滚动加载三种分页类型，自动去重，避免无限循环。

核心代码：

import requests
from bs4 import BeautifulSoup
from urllib.parse import urlparse, parse_qs
import hashlib
import random
import time

class PaginationCrawler:
    def __init__(self, base_url, pagination_type="page"):
        self.base_url = base_url
        self.pagination_type = pagination_type  # page/offset/cursor
        self.crawled = set()  # 去重集合
        self.max_pages = 100  # 最大爬取页数（防无限循环）
        self.max_data = 10000  # 最大数据量
        self.crawled_count = 0  # 已爬数据量
    
    def get_page_content(self, url):
        """获取页面内容（处理请求异常）"""
        headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}
        try:
            time.sleep(random.uniform(1, 2))  # 控制速度
            response = requests.get(url, headers=headers, timeout=10)
            response.raise_for_status()
            return response.text
        except Exception as e:
            print(f"请求失败：{e}")
            return None
    
    def parse_data(self, content):
        """解析数据（需根据实际页面重写）"""
        soup = BeautifulSoup(content, "lxml")
        items = soup.select("div.item")
        self.crawled_count += len(items)
        return items
    
    def get_next_page_url(self, current_page, content=None):
        """根据分页类型构造下一页URL"""
        if self.pagination_type == "page":
            # 静态页码分页
            parsed = urlparse(self.base_url)
            params = parse_qs(parsed.query)
            params["page"] = [str(current_page)]
            new_query = "&".join([f"{k}={v[0]}" for k, v in params.items()])
            return parsed._replace(query=new_query).geturl()
        elif self.pagination_type == "offset":
            # 偏移量分页
            offset = (current_page - 1) * 20
            return f"{self.base_url}?offset={offset}&limit=20"
        elif self.pagination_type == "cursor":
            # 滚动分页（从content中提取cursor，需重写）
            # 示例：从HTML中提取next_cursor
            soup = BeautifulSoup(content, "lxml")
            next_cursor = soup.select_one("input[name='next_cursor']").get("value", "")
            return f"{self.base_url}?cursor={next_cursor}" if next_cursor else None
    
    def is_duplicate(self, url, content):
        """双重去重：URL+数据MD5"""
        if url in self.crawled:
            return True
        # 计算核心数据MD5
        soup = BeautifulSoup(content, "lxml")
        core_data = soup.select_one("div.list-container").get_text(strip=True) if soup.select_one("div.list-container") else ""
        data_md5 = hashlib.md5(core_data.encode("utf-8")).hexdigest()
        if data_md5 in self.crawled:
            return True
        self.crawled.add(url)
        self.crawled.add(data_md5)
        return False
    
    def crawl(self):
        current_page = 1
        while current_page <= self.max_pages and self.crawled_count < self.max_data:
            # 1. 构造当前页URL
            current_url = self.get_next_page_url(current_page)
            if not current_url:
                print("无下一页URL，终止")
                break
            print(f"爬取第{current_page}页：{current_url}")
            
            # 2. 获取页面内容
            content = self.get_page_content(current_url)
            if not content:
                current_page += 1
                continue
            
            # 3. 去重校验
            if self.is_duplicate(current_url, content):
                print("重复页面，终止爬取")
                break
            
            # 4. 解析数据
            items = self.parse_data(content)
            print(f"提取{len(items)}条数据，累计{self.crawled_count}条")
            
            # 5. 终止条件校验
            if not items:
                print("当前页无数据，终止")
                break
            
            current_page += 1

# 使用示例（静态页码分页）
crawler = PaginationCrawler("https://example.com/products?page=1", pagination_type="page")
crawler.crawl()

七、总结

避免爬虫无限循环的核心是“明确分页规则+多重去重+刚性终止条件”：

先通过开发者工具分析分页类型（URL参数/动态接口/游标），再构造下一页链接；去重优先选择“URL+数据ID/MD5”双重校验，避免单一去重策略的漏洞；终止条件必须“多重兜底”（数据为空+页数阈值+数据量阈值），防止因接口异常导致无限循环；动态分页（AJAX/滚动）优先用Requests直接调用接口，比Selenium更高效、更易控制。

实际开发中，需根据网站分页的具体实现灵活调整策略——重点关注“分页参数是否唯一”“是否有明确的终止标识”“是否存在虚假分页”三个核心问题，即可彻底解决无限循环问题。

如果在实战中遇到复杂分页场景（如加密分页参数、动态生成cursor）、去重性能瓶颈等问题，欢迎留言交流！