# 客户端轮换

ISP IP 永久稳定——没有 `session` 字段、没有网关轮换。需要按请求换 IP 时，由你的客户端挑。本页给出常见模式。

## 轮询

最简单：按清单顺序循环。

```python
import itertools, requests

with open("isp.txt") as f:
    lines = [l.strip() for l in f if l.strip()]

cycle = itertools.cycle(lines)

def get(url):
    ip, port, user, password = next(cycle).split(":", 3)
    proxy = f"http://{user}:{password}@{ip}:{port}"
    return requests.get(url, proxies={"http": proxy, "https": proxy}, timeout=30)
```

优点：简单。缺点：失效 IP 仍会被选中。

## 轮询 + 健康追踪

记录失败次数，在冷却窗口内跳过失效 IP：

```python
import time, itertools, requests
from collections import defaultdict

class IPPool:
    def __init__(self, lines, cooldown=300):
        self.lines = lines
        self.cooldown = cooldown
        self.bad_until = defaultdict(float)
        self._iter = itertools.cycle(lines)

    def next_alive(self):
        for _ in range(len(self.lines)):
            line = next(self._iter)
            if self.bad_until[line] < time.time():
                return line
        raise RuntimeError("All IPs are in cool-down")

    def mark_bad(self, line):
        self.bad_until[line] = time.time() + self.cooldown

pool = IPPool(open("isp.txt").read().splitlines())

def get(url):
    line = pool.next_alive()
    ip, port, user, password = line.split(":", 3)
    proxy = f"http://{user}:{password}@{ip}:{port}"
    try:
        r = requests.get(url, proxies={"http": proxy, "https": proxy}, timeout=30)
        if r.status_code >= 500:
            pool.mark_bad(line)
        return r
    except requests.RequestException:
        pool.mark_bad(line)
        raise
```

## 按任务粘性

任务单元需要稳定 IP（登录流程、多步购买），把单个 IP 锁定到任务：

```python
import hashlib

def ip_for_task(task_id, lines):
    h = int(hashlib.sha256(task_id.encode()).hexdigest(), 16)
    return lines[h % len(lines)]
```

一致性哈希让重启或重跑同一 `task_id` 总落到同一 IP。

## 并发上限

单 ISP IP 吞吐很高（1 Gbps+），但多数目标站点会按 IP 限流。**对单一目标站，每 IP 起步建议 5 个并发**，分摊到整批后线性扩展。

## 跨进程协同

多 worker 共用一批 IP 时，用共享存储（Redis、etcd）协调轮换与健康追踪。Redis 简单版：

```python
import redis, time, json

r = redis.Redis()

def pick(batch_key):
    lines = json.loads(r.get(f"isp:{batch_key}:lines"))
    idx = r.incr(f"isp:{batch_key}:counter") % len(lines)
    return lines[idx]
```

健康追踪可用按 `cooldown_until` 排序的 sorted set，跳过 `now()` 之后的条目。

## 不该轮换的情况

如果使用场景是账号绑定（登录抓取、发帖、监控已登录的仪表盘），**不要轮换**。每账号锁定一个 IP。使用 ISP 的初衷就是 IP 稳定，轮换会让它失效。

## 现成的轮换工具

* **反检测浏览器**（AdsPower、Multilogin、GoLogin）— 每个浏览器画像分配不同 ISP IP
* **Proxifier** — OS 层按导入清单轮询
* **Scrapy + scrapy-rotating-proxies** — 中间件读清单做轮换
* **Apify SDK proxy configuration** — 传入清单后自动轮换

多数生产栈用爬虫框架自带的轮换器就够了，上面的代码用于需要自定义逻辑时。


---

# Agent Instructions: Querying This Documentation

If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter:

```
GET https://docs.helodata.com/helodata-zh/chan-pin/overview-2/rotation.md?ask=<question>
```

The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.