圆海博客-探寻心灵的宁静

您现在的位置是:首页 > 博客 > 正文

博客

python动态ip代理教程:requests库轮换ip实战代码示例

2026-09-03 21:03:08博客
做数据采集、价格监控或社交媒体运营时,你是不是也遇到过这样的困境:请求了不到几百条数据,目标网站就把你的 IP 封了?换一台服务器接着跑,过不了半天又被封。
问题的根源在于—……

做数据采集、价格监控或社交媒体运营时,你是不是也遇到过这样的困境:请求了不到几百条数据,目标网站就把你的 IP 封了?换一台服务器接着跑,过不了半天又被封。

问题的根源在于——同一个出口 IP 短时间内大量请求,触发了目标网站的反爬策略。解决办法其实很直接:让每一次请求都从一个不同的 IP 出去。这就是”动态 IP 代理“的核心价值。

这篇文章不聊虚的,直接上 Python requests 库 的实战代码,从最基础的代理配置,到完整的 IP 轮换批量采集脚本,一步步带你跑通。看完你就能把动态 IP 代理集成到自己的项目里。

一、动态 IP 代理到底解决了什么问题

先花 30 秒搞清楚”动态”和”静态”的区别,后面写代码时你就知道为什么必须用动态的。

静态 IP 代理:你拿到一个固定地址,每次请求都从同一个出口出去。适合需要”看起来像同一个用户”的场景,比如固定账号登录。但缺点是——请求量一大,这个 IP 很快就会被目标网站标记、限速甚至封禁。

动态 IP 代理:每次请求(或每隔固定时间)自动切换到一个新的出口 IP。IP 池通常有数万甚至数十万个,单个 IP 的使用频率极低,目标网站很难在短期内积累足够的”坏行为”记录来封你。

简单总结:

对比维度 动态 IP 代理 静态 IP 代理
IP 是否变化 每次请求或定时切换 固定不变
抗封禁能力 强(IP 分散,单 IP 压力小) 弱(单 IP 持续暴露)
适用场景 批量采集、大规模监控 固定账号操作、API 调用
IP 池规模 数万 ~ 数十万 单个或少数几个

对于绝大多数 Python 采集项目来说,动态 IP 是刚需。下面开始写代码。

二、requests 库配置代理:最基础的写法

requests 库原生支持 HTTP 代理,只需要在请求时传入 proxies 参数即可。这是所有后续代码的地基。

import requests

# 代理地址格式:协议://用户名:密码@IP:端口
proxy = "http://user123:pass456@203.0.113.10:8080"

proxies = {
    "http": proxy,
    "https": proxy
}

# 带代理的请求
resp = requests.get(
    "https://httpbin.org/ip",
    proxies=proxies,
    timeout=10
)

print(resp.json())
# 输出:{"origin": "203.0.113.10"}  ← 这就是你的代理出口IP

就这几行,你的请求已经不再从本机 IP 出去了,而是经过代理服务器转发。但注意——这里用的是一个固定 IP,本质上还是静态代理的用法。要实现”动态轮换”,还需要额外处理。

三、动态轮换 IP:三种实战模式

动态 IP 的”轮换”有几种常见实现方式,根据代理服务商提供的接口不同,代码写法也不一样。下面三种模式覆盖了 90% 的使用场景。

模式 A:每次请求自动换 IP(推荐)

部分动态代理服务商支持”sticky session”机制——你只需设置一个 session 超时时间(比如 60 秒),在这个时间窗口内所有请求走同一个 IP,超时后自动切换。代码上你什么都不用改,代理侧自动处理:

import requests

# 带 session 参数的代理地址(以光络云为例,具体参数看服务商文档)
proxy_url = "http://user:pass@proxy.example.com:port?session=60"

proxies = {
    "http": proxy_url,
    "https": proxy_url
}

# 循环请求,每 60 秒自动换一个出口 IP
for i in range(100):
    try:
        r = requests.get("https://httpbin.org/ip", proxies=proxies, timeout=10)
        print(f"第 {i+1} 次 → 出口IP: {r.json()['origin']}")
    except requests.RequestException as e:
        print(f"第 {i+1} 次 → 请求失败: {e}")

这种模式最省心,适合请求间隔比较均匀的场景。

模式 B:手动从 IP 池拉取,逐个轮换

如果你的代理服务商提供的是一个 IP 列表接口(每次调用返回一个新的可用 IP),你可以自己控制轮换节奏:

import requests
import time

def get_new_ip():
    """从代理服务商接口获取一个新的出口IP"""
    api = "http://api.proxy-provider.com/get-ip?user=xxx&pass=xxx"
    r = requests.get(api, timeout=5)
    return r.text.strip()  # 返回 "203.0.113.42:8080"

def fetch_with_rotation(url, total=50):
    """批量请求,每个IP最多用3次就换"""
    results = []
    ip = get_new_ip()
    use_count = 0

    for i in range(total):
        if use_count >= 3:
            ip = get_new_ip()
            use_count = 0
            print(f"  → 切换新IP: {ip}")

        proxies = {
            "http":  f"http://{ip}",
            "https": f"http://{ip}"
        }

        try:
            r = requests.get(url, proxies=proxies, timeout=15)
            r.raise_for_status()
            results.append(r.text)
            use_count += 1
        except requests.RequestException as e:
            print(f"  第{i+1}次请求失败: {e},立即换IP")
            ip = get_new_ip()
            use_count = 0
            # 重试当前请求
            try:
                r = requests.get(url, proxies=proxies, timeout=15)
                results.append(r.text)
            except:
                results.append(None)

    return results

data = fetch_with_rotation("https://httpbin.org/html", total=50)

这种模式灵活度最高,你可以精确控制每个 IP 的使用次数、切换频率、失败重试策略

模式 C:并发请求 + 线程池 + IP 轮换

当请求量上来之后,串行太慢了。用 ThreadPoolExecutor 并发跑,每个线程分配一个独立 IP:

import requests
from concurrent.futures import ThreadPoolExecutor, as_completed

def worker(task_id, ip):
    """每个线程用独立IP发请求"""
    proxies = {"http": f"http://{ip}", "https": f"http://{ip}"}
    try:
        r = requests.get(
            "https://httpbin.org/ip",
            proxies=proxies,
            timeout=10
        )
        return task_id, r.json().get("origin"), None
    except Exception as e:
        return task_id, None, str(e)

# 准备10个IP
ip_pool = [f"203.0.113.{i}:8080" for i in range(1, 11)]
tasks = [(i, ip_pool[i % len(ip_pool)]) for i in range(30)]

with ThreadPoolExecutor(max_workers=10) as pool:
    futures = {pool.submit(worker, tid, ip): tid for tid, ip in tasks}
    for f in as_completed(futures):
        tid, origin_ip, err = f.result()
        if err:
            print(f"任务{tid} 失败: {err}")
        else:
            print(f"任务{tid} 成功 → 出口: {origin_ip}")

30 个请求、10 个 IP、10 个线程并发,几秒就跑完,而且每个请求走不同出口,目标网站看到的只是”10 个不同用户各发了几条请求”,完全不像爬虫。

四、生产环境必须加的防御代码

上面是”能跑”的版本。真正上生产环境,你至少还要处理这几个问题:

1. 超时与重试:代理链路比直连多了一跳,网络波动时更容易超时。设置合理的 timeout,失败后换 IP 重试,而不是直接报错退出。

2. 请求头伪装:动态 IP 只解决了”地址”问题,你的 User-AgentAccept-Language 等请求头如果还是 Python-requests 的默认值,一样会被识别。建议每次请求随机切换 UA:

import random

UA_LIST = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15",
    "Mozilla/5.0 (X11; Linux x86_64; rv:109.0) Gecko/20100101 Firefox/119.0",
    "Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15",
]

headers = {
    "User-Agent": random.choice(UA_LIST),
    "Accept": "text/html,application/xhtml+xml",
    "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
}

r = requests.get(url, headers=headers, proxies=proxies, timeout=15)

3. 请求间隔:即便 IP 在轮换,同一个 IP 上连续打几十个请求也不自然。在每次请求之间加一个 time.sleep(random.uniform(0.5, 2)),模拟人类操作节奏。

4. 日志与监控:记录每次请求的出口 IP、状态码、耗时。一旦某个 IP 连续返回 403/429,立刻把它标记为”坏 IP”,从池中剔除,避免反复踩坑。

五、如何选择动态 IP 代理服务商

代码写好了,但代理 IP 本身的质量直接决定你的采集成功率。选服务商时重点看这几点:

IP 池规模与质量:池子越大,单个 IP 被复用的概率越低。住宅 IP 比机房 IP 更难被识别,如果你的目标网站反爬比较严,优先选住宅级 IP。

轮换机制是否灵活:支持 session 超时控制?支持手动指定 IP?支持按地域筛选?这些决定了你能不能精确匹配业务需求。

在线率与延迟:动态 IP 池里难免有”死 IP”。服务商的在线率应该稳定在 99% 以上,平均延迟控制在 50ms 以内,否则你的请求大量超时,采集效率会大打折扣。

是否提供 API 接口:像上面模式 B 那样手动拉 IP、管理 IP 生命周期,需要服务商提供稳定的 API。如果只能给你一个固定入口地址,灵活性会差很多。

光络云提供国内和海外动态 IP 代理服务,IP 池覆盖住宅级和机房级,支持 session 控制、地域筛选和 API 管理,平均延迟和在线率都表现稳定。需要注意的一点是:光络云的代理 IP 需要你的服务器或本地环境具备海外网络出口才能正常连接(TikTok 专线产品除外)。如果你的服务器在国内且没有海外网络环境,这一点要提前确认。

六、完整实战:用动态 IP 采集商品价格

把前面所有知识点串起来,下面是一个可以直接改改就用的完整示例——采集某电商网站 200 个商品的价格,每个 IP 最多用 5 次,失败自动换 IP 重试:

import requests
import time
import random
import logging
from concurrent.futures import ThreadPoolExecutor, as_completed

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")
logger = logging.getLogger(__name__)

# ===== 配置区 =====
PROXY_API    = "http://api.ipipgo.com/get-ip?user=YOUR_USER&pass=YOUR_PASS"
MAX_USE_PER_IP = 5       # 每个IP最多用几次
MAX_RETRIES    = 3       # 单条请求最大重试次数
WORKERS        = 8        # 并发线程数
UA_LIST = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120.0",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) Safari/605.1.15",
    "Mozilla/5.0 (X11; Linux x86_64; rv:109.0) Firefox/119.0",
]

# ===== IP 管理 =====
def fetch_proxy():
    r = requests.get(PROXY_API, timeout=5)
    return r.text.strip()

class ProxyPool:
    def __init__(self):
        self.ip = fetch_proxy()
        self.count = 0
        self.bad_ips = set()

    def get(self):
        if self.count >= MAX_USE_PER_IP or self.ip in self.bad_ips:
            self.ip = fetch_proxy()
            self.count = 0
            logger.info(f"切换新IP: {self.ip}")
        self.count += 1
        return f"http://{self.ip}"

    def mark_bad(self, ip):
        self.bad_ips.add(ip)
        logger.warning(f"标记坏IP: {ip}")

pool = ProxyPool()

# ===== 单条请求 =====
def scrape_one(url, proxy_pool):
    for attempt in range(MAX_RETRIES):
        proxy = proxy_pool.get()
        proxies = {"http": proxy, "https": proxy}
        headers = {
            "User-Agent": random.choice(UA_LIST),
            "Accept": "text/html",
            "Accept-Language": "zh-CN,zh;q=0.9",
        }
        try:
            r = requests.get(url, proxies=proxies, headers=headers, timeout=15)
            if r.status_code == 200:
                return url, r.text
            elif r.status_code in (403, 429):
                proxy_pool.mark_bad(proxy.split("//")[-1])
            time.sleep(random.uniform(1, 3))
        except requests.RequestException as e:
            logger.warning(f"请求异常 (第{attempt+1}次): {e}")
            time.sleep(1)
    return url, None

# ===== 主流程 =====
def main():
    urls = [f"https://example.com/product/{i}" for i in range(1, 201)]
    results = {}

    with ThreadPoolExecutor(max_workers=WORKERS) as executor:
        futures = {
            executor.submit(scrape_one, url, pool): url
            for url in urls
        }
        done = 0
        for f in as_completed(futures):
            url, html = f.result()
            done += 1
            if html:
                results[url] = html
                # 在这里解析价格,比如用 BeautifulSoup
            else:
                logger.error(f"最终失败: {url}")
            if done % 20 == 0:
                logger.info(f"进度: {done}/200")

    logger.info(f"完成! 成功 {len(results)}/200")
    # 保存结果...
    return results

if __name__ == "__main__":
    main()

PROXY_API 换成你的光络云代理接口地址,urls 换成你的目标列表,这个脚本就能直接跑。200 个商品、8 个线程并发、每个 IP 用 5 次就换——整个采集过程大概 2-3 分钟,成功率通常能到 95% 以上

常见问题

Q: 动态 IP 代理和 VPN 有什么区别?
两者都能隐藏你的真实 IP,但原理和用途完全不同。动态 IP 代理是在 HTTP 层面转发请求,可以精确控制出口 IP、支持并发和 API 管理,适合程序化采集。VPN 是在网络层加密隧道,整个系统流量都走 VPN,适合个人隐私保护,不适合批量采集场景。

Q: 为什么我的请求经过代理后速度变慢了?
代理多了一跳转发,延迟通常会增加 20-50ms。如果明显变慢(比如超过 200ms),大概率是代理服务商的线路质量有问题,或者你选的出口节点离目标网站太远。可以试试按地域筛选代理节点,选离目标网站最近的区域。

Q: 光络云的代理 IP 在国内服务器能直接用吗?
光络云的代理 IP 需要你的运行环境具备海外网络出口才能正常连接使用(TikTok 专线产品除外)。如果你的服务器部署在国内且没有海外网络环境,需要先解决网络出口问题,或者考虑使用光络云的 TikTok 专线产品。

Q: 一个 IP 用多少次比较安全?
没有统一标准,取决于目标网站的反爬策略。一般建议每个 IP 控制在 3-10 次请求之间。如果目标网站比较敏感(比如金融、社交类),建议每个 IP 只用 1-2 次就换。上面代码里的 MAX_USE_PER_IP = 5 是一个比较通用的起点,你可以根据实际封禁率调整。

Q: 动态 IP 代理能绕过验证码吗?
不能。动态 IP 解决的是”IP 被识别和封禁”的问题,但验证码(滑块、图形、短信)是针对”行为”的验证,跟 IP 无关。如果你的目标网站有验证码,需要额外集成打码服务或无头浏览器来模拟操作,动态 IP 只是整个反反爬方案中的一个环节。