芹菜，RQ，BullMQ：带代理和IP限制的任务队列

当解析器或机器人每分钟从一个IP发送数百个请求时，服务器会迅速启用速率限制，接下来就是验证码或封禁。解决方案是任务队列，它在工作者和代理之间分配请求。在本文中，我们将探讨Celery、RQ和BullMQ中这种架构的工作原理，以及如何合理管理每个IP地址的请求限制。

使用代理时为什么需要任务队列

如果任务是从Wildberries抓取50,000个商品卡片或检查10,000个Instagram账户的状态，顺序发送请求将耗费数小时，而没有控制的并行发送将迅速导致IP被封。任务队列（task queue）解决了这两个问题：它将大量工作分解为小任务，将它们分配给多个工作者，并允许将每个工作者绑定到单独的代理。

架构通常如下：消息代理（Redis或RabbitMQ）存储任务队列，工作者从队列中获取任务并执行，每个工作者使用自己的代理或从池中轮换代理。这使得即使请求总量巨大，每个IP的负载也能保持在安全范围内。

这种方法不仅被解析器的开发者使用，还被自动化检查Facebook Ads广告账户、监控Ozon价格或同时从数十个网站收集数据的团队使用。如果没有队列和合理的IP分配，任何这样的系统在工作第一天就注定会被封锁。

单个IP的请求限制问题

大多数网站和API都有防止过载的保护措施。它可以以不同的方式表现出来：服务器返回HTTP 429（请求过多），显示验证码，暂时封禁IP 10-60分钟，或将其永久加入黑名单。不同平台的限制差异很大——例如，Wildberries API允许从一个地址每秒发送有限数量的请求，而Instagram在15-20个连续操作后就会对可疑活动做出反应。

建立队列时的主要错误是通过同一个IP启动所有工作者。即使任务在20个工作者之间分配，但如果它们都使用同一个代理或根本不使用代理，该IP的限制也会像顺序执行时一样迅速到来。因此，工作者的数量应与代理池中可用的IP地址数量相对应。

对于高频率请求的单个域，最好使用数据中心代理——它们提供低延迟，适合大规模并行抓取，其中速度比伪装成普通用户更为重要。如果目标是绕过更严格的保护，例如市场或社交网络的反机器人系统，则需要看起来像普通用户的IP，即住宅或移动IP。

Celery：代码中的架构和代理轮换

Celery是Python生态系统中最成熟和流行的任务队列。它运行在代理之上（通常是Redis或RabbitMQ），支持灵活的并发性、重试和任务优先级设置。为了方便代理轮换，可以将IP池存储在Redis中，并在任务入队时为每个任务分配代理。

from celery import Celery
import requests
import redis
import random

app = Celery('tasks', broker='redis://localhost:6379/0')
redis_client = redis.Redis(host='localhost', port=6379, db=1)

PROXY_POOL = [
    "http://user:[email protected]:8000",
    "http://user:[email protected]:8000",
    "http://user:[email protected]:8000",
]

def get_proxy():
    # 通过Redis进行轮询，以避免对同一个IP的冲击
    idx = redis_client.incr("proxy_cursor") % len(PROXY_POOL)
    return PROXY_POOL[idx]

@app.task(bind=True, max_retries=3, default_retry_delay=30)
def fetch_url(self, url):
    proxy = get_proxy()
    try:
        response = requests.get(
            url,
            proxies={"http": proxy, "https": proxy},
            timeout=15
        )
        if response.status_code == 429:
            raise self.retry(exc=Exception("Rate limited"))
        return response.status_code
    except requests.exceptions.RequestException as exc:
        raise self.retry(exc=exc)

关键点是函数get_proxy()，它通过Redis计数器循环返回池中的下一个代理。这确保了负载在所有IP之间均匀分配，而不是集中在一个IP上。参数max_retries和对429代码的处理允许任务在当前IP受到限制时自动重新发送使用其他代理。

为了控制工作者级别的并发性，Celery允许通过启动工作者时的--concurrency参数限制同时任务的数量，以及通过不同优先级的队列（例如，针对具有严格速率限制的慢任务的单独队列）。

RQ (Redis Queue)：简单的替代方案

RQ是一个比Celery更轻量的库，没有RabbitMQ代理，配置最小。它适合于小型项目，不需要复杂的任务路由，只需要一个带工作者的队列，并能够将代理绑定到任务上。

from redis import Redis
from rq import Queue
import requests
import itertools

redis_conn = Redis()
q = Queue(connection=redis_conn)

PROXY_POOL = [
    "http://user:[email protected]:8000",
    "http://user:[email protected]:8000",
]
proxy_cycle = itertools.cycle(PROXY_POOL)

def fetch_with_proxy(url):
    proxy = next(proxy_cycle)
    response = requests.get(
        url,
        proxies={"http": proxy, "https": proxy},
        timeout=15
    )
    if response.status_code == 429:
        # 带延迟的重新入队
        q.enqueue_in(60, fetch_with_proxy, url)
        return None
    return response.status_code

# 将任务入队
urls = ["https://example.com/page1", "https://example.com/page2"]
for url in urls:
    q.enqueue(fetch_with_proxy, url)

在RQ中，通过itertools.cycle的轮换在工作者进程内部进行——这比Redis计数器简单，但在多个并行工作者的情况下扩展性较差，因为每个工作者都从头开始循环。对于1-5个工作者的小型项目，这已经足够，而对于更大的项目，最好回到像Celery示例中那样的Redis集中计数器。

RQ非常适合已经在项目中使用Redis的开发者，并且不想引入单独的消息代理。缺点是与Celery相比，内置的优先级和监控机制较少，因此对于每天处理数十万任务的生产系统，通常选择Celery。

BullMQ：带IP限制的Node.js队列

BullMQ是Node.js生态系统中任务队列的标准，构建在Redis之上，并具有内置的速率限制器，这对于控制IP请求限制特别方便。可以在队列级别直接指定限制器，而不是手动计算请求。

const { Queue, Worker } = require('bullmq');
const axios = require('axios');

const connection = { host: 'localhost', port: 6379 };

const proxyPool = [
  'http://user:[email protected]:8000',
  'http://user:[email protected]:8000',
  'http://user:[email protected]:8000',
];

const queue = new Queue('fetchQueue', { connection });

// 限制器：每秒最多5个任务
const worker = new Worker('fetchQueue', async job => {
  const proxy = proxyPool[job.id % proxyPool.length];
  try {
    const response = await axios.get(job.data.url, {
      proxy: false,
      httpsAgent: new (require('https-proxy-agent'))(proxy),
      timeout: 15000,
    });
    return response.status;
  } catch (err) {
    if (err.response && err.response.status === 429) {
      throw new Error('Rate limited, retrying with another proxy');
    }
    throw err;
  }
}, {
  connection,
  limiter: { max: 5, duration: 1000 },
  concurrency: proxyPool.length,
});

for (let i = 0; i < 100; i++) {
  queue.add('fetch', { url: `https://example.com/item/${i}` }, {
    attempts: 3,
    backoff: { type: 'exponential', delay: 5000 },
  });
}

BullMQ中的limiter参数是一个内置机制，它限制在一段时间内处理的任务总数，这对于遵守全系统的请求限制非常方便。而concurrency参数等于代理池的大小，确保同时运行的工作者数量正好与可用的IP数量相同。

这种方法通常被编写自己的机器人以自动化TikTok或Telegram操作的团队使用——BullMQ与其他Node.js栈良好集成，并且不需要为了一个任务而切换到Python。

Celery、RQ与BullMQ的比较

参数	Celery	RQ	BullMQ
语言	Python	Python	Node.js
代理	Redis / RabbitMQ	仅Redis	仅Redis
内置速率限制器	没有（手动编写）	没有	是，开箱即用
入门门槛	中等	低	中等
适合于	大型生产系统	小型项目，原型	Node.js栈，机器人，自动化

IP负载分配策略

上述示例中使用的轮询是最简单的策略，但并不总是最优。以下是一些根据任务值得考虑的选项：

轮询——每个后续请求通过下一个代理循环进行。当所有IP相等且限制相同时适用。
粘性会话——一个代理绑定到特定账户或会话，持续整个操作时间。对于社交媒体和市场，IP在会话中间的更换看起来可疑，因此至关重要。
加权分配——部分请求被发送到更昂贵和可靠的IP（例如，移动IP），部分请求被发送到便宜的数据中心IP，具体取决于正在处理的域。
自适应节流——如果429错误或验证码的比例上升，则自动降低任务发送速度，并在错误消失时恢复。

对于需要伪装成真实用户的任务——例如，通过反检测浏览器（如Dolphin Anty或AdsPower）加热和检查Instagram或TikTok账户——最好使用带有移动代理的粘性会话：它们更少受到反机器人系统的怀疑，并表现得像普通的移动互联网用户。

对于从网站进行大规模数据收集的任务，速度和数量比算法的信任更重要，使用带有住宅代理的轮询在稳定性和价格之间提供了平衡：IP看起来像真实用户，但可以频繁更换而不影响特定会话的风险。

监控429错误和封禁

没有监控的任务队列就像一个黑箱：很难理解为什么部分任务失败以及哪些IP已经受到怀疑。值得跟踪的最小指标集：

每个代理的429和403错误响应比例
平均响应时间——延迟的增加通常会在封禁之前发生
每个任务的重试次数（retries）
实时任务队列的长度——如果它在增长，工作者无法应对

在Celery中，可以使用Flower——用于实时监控任务和工作者的Web界面。在BullMQ中，有Bull Board提供类似功能。RQ有RQ Dashboard。这三种工具显示任务状态、失败尝试的数量，并允许快速了解哪个代理应暂时从轮换中排除。

一个好的做法是在几次连续的429错误后，将代理自动标记为“隔离”15-30分钟，并在成功的测试请求后才将其返回到池中。这可以防止被封的IP继续接收任务并浪费工作者的时间。

结论

任务队列解决了请求的扩展问题，但本身并不能保护IP限制——为此需要在Celery、RQ或BullMQ内部进行深思熟虑的代理轮换。工具的选择取决于技术栈：Python项目通常选择Celery用于生产，RQ用于简单任务，而Node.js开发者则选择BullMQ及其内置的速率限制器。

无论选择哪种队列，结果都在很大程度上取决于代理池的质量。如果任务是快速的大规模抓取，请考虑使用数据中心代理以提高速度，而对于社交媒体和市场，忠于反机器人系统的住宅或移动IP更为合适——它们降低了429错误和封禁的比例，这直接减少了重试次数并加快了整个任务队列的执行。