セロリ、RQ、BullMQ：プロキシとIP制限付きのタスクキュー

パーサーやボットが1つのIPから毎分数百のリクエストを送信すると、サーバーはすぐにレートリミットを適用し、その後はキャプチャまたはバンが待っています。解決策は、リクエストをワーカーとプロキシ間で分配するタスクキューです。この記事では、セロリ、RQ、BullMQにおけるこのようなアーキテクチャがどのように機能するか、各IPアドレスに対するリクエスト制限を適切に管理する方法について説明します。

プロキシを使用する際のタスクキューの必要性

もしタスクがWildberriesから50,000の商品カードをスクレイピングするか、10,000のInstagramアカウントのステータスを確認することであれば、リクエストを順番に実行すると数時間かかり、制御なしで並行実行するとすぐにIPがバンされます。タスクキューはこの2つの問題を解決します。大きな作業を小さなタスクに分割し、それらを複数のワーカーに分配し、各ワーカーに個別のプロキシを割り当てることができます。

アーキテクチャは通常次のようになります: メッセージブローカー（RedisまたはRabbitMQ）がタスクキューを保持し、ワーカーがキューからタスクを取得して実行します。各ワーカーは自分のプロキシを使用するか、プールからプロキシをローテーションします。これにより、全体のリクエスト量が巨大であっても、各IPへの負荷を安全な範囲内に保つことができます。

このアプローチは、パーサーの開発者だけでなく、Facebook Adsの広告アカウントのチェックを自動化したり、Ozonの価格を監視したり、同時に数十のサイトからデータを収集するチームにも適用されます。キューとIPによる適切な分配がなければ、そのようなシステムは初日からブロックされる運命にあります。

1つのIPに対するリクエスト制限の問題

ほとんどのウェブサイトやAPIには過剰な負荷からの保護があります。これは様々な形で現れることがあります: サーバーがHTTP 429（Too Many Requests）を返したり、キャプチャを表示したり、IPを10〜60分間一時的にバンしたり、永遠にブラックリストに追加したりします。異なるプラットフォームでは制限が大きく異なります — 例えば、Wildberries APIは1つのアドレスからのリクエストを制限された数だけ受け入れ、Instagramは15〜20の連続アクションの後に疑わしい活動に反応します。

キューを構築する際の主な誤りは、すべてのワーカーを同じIPを介して起動することです。タスクが20のワーカーに分配されていても、すべてが1つのプロキシを使用するか、全く使用しない場合、そのIPに対する制限は順次実行と同じくらい早く発生します。したがって、ワーカーの数はプロキシプール内の利用可能なIPアドレスの数に関連付ける必要があります。

1つのドメインに対するリクエスト頻度が高いタスクには、データセンターのプロキシを使用するのが最適です — これにより低遅延が得られ、大規模な並行スクレイピングに適しています。速度が重要であり、通常のユーザーとしての偽装が必要ない場合に最適です。もし目標がマーケットプレイスやソーシャルネットワークのアンチボットシステムのような厳しい保護を回避することであれば、通常のユーザーに似たIP、つまりレジデンシャルまたはモバイルが必要です。

セロリ: アーキテクチャとコード内のプロキシローテーション

セロリはPythonエコシステムで最も成熟した人気のあるタスクキューです。これはブローカー（通常はRedisまたはRabbitMQ）の上で動作し、競合性、再試行、タスクの優先順位の柔軟な設定をサポートします。プロキシのローテーションには、RedisにIPプールを保存し、キューに追加する際に各タスクにプロキシを割り当てるのが便利です。

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を介してラウンドロビン、1つの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("レート制限"))
        return response.status_code
    except requests.exceptions.RequestException as exc:
        raise self.retry(exc=exc)

重要なポイントは、get_proxy()関数で、Redisカウンターを介してプールから次のプロキシをラウンドロビンで取得します。これにより、負荷がすべてのIPに均等に分配され、1つに集中しないことが保証されます。max_retriesパラメータと429コードの処理により、現在のIPが制限を受けた場合にタスクが自動的に別のプロキシで再送信されます。

セロリでは、ワーカーの競合性を制御するために、ワーカー起動時に--concurrencyパラメータを使用して同時タスクの数を制限したり、異なる優先順位のキューを使用したりできます（例えば、厳しいレートリミットのある遅いタスク用の別のキュー）。

RQ (Redis Queue): シンプルな代替手段

RQは、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のワーカーを持つ小規模なプロジェクトには十分ですが、より大規模なプロジェクトには、セロリの例のようにRedis内の中央集計カウンターに戻る方が良いでしょう。

RQは、すでにプロジェクトでRedisを使用している開発者に適していますが、別のメッセージブローカーを立ち上げたくない場合に便利です。欠点は、セロリに比べて組み込みの優先順位付けや監視機能が少ないため、1日に数十万のタスクを処理するプロダクションシステムでは、セロリが選ばれることが多いです。

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 });

// リミッター: キューあたり1秒間に最大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('レート制限、別のプロキシで再試行中');
    }
    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に切り替える必要がありません。

セロリ、RQ、BullMQの比較

パラメータ	セロリ	RQ	BullMQ
言語	Python	Python	Node.js
ブローカー	Redis / RabbitMQ	Redisのみ	Redisのみ
組み込みのレートリミッター	なし（手動で記述）	なし	あり、デフォルトで
参入障壁	中程度	低い	中程度
適している用途	大規模なプロダクションシステム	小規模なプロジェクト、プロトタイプ	Node.jsスタック、ボット、自動化

IPによる負荷分散の戦略

上記の例で使用されているラウンドロビンは最もシンプルな戦略ですが、常に最適というわけではありません。タスクに応じて検討すべきいくつかのオプションがあります:

ラウンドロビン — 次のリクエストは次のプロキシを介して回ります。すべてのIPが同等で、制限が同じ場合に適しています。
スティッキーセッション — 1つのプロキシが特定のアカウントまたはセッションに固定され、作業中ずっとそのままです。ソーシャルメディアやマーケットプレイスでは、セッションの途中でIPを変更することが疑わしく見えるため、重要です。
ウェイテッドディストリビューション — より高価で信頼性の高いIP（例えば、モバイル）に一部のリクエストを向け、安価なデータセンターのものに一部を向けることで、処理するドメインに応じて調整します。
適応的スロットリング — エラー429やキャプチャの割合が増加した場合、自動的にタスクの送信速度を減少させ、エラーが消えたときに回復します。

生のユーザーとしての偽装が重要なタスク、例えばInstagramやTikTokのアカウントの温めや確認には、モバイルプロキシを使用したスティッキーセッションが最適です。これにより、アンチボットシステムの疑いを避け、通常のモバイルインターネットユーザーのように振る舞います。

スピードとボリュームが重要で、アルゴリズムの信頼性が重要でないウェブサイトからのデータ収集には、レジデンシャルプロキシを使用したラウンドロビンが、安定性と価格のバランスを提供します。IPは本物のユーザーのように見えますが、特定のセッションにリスクを与えることなく頻繁に変更できます。

エラー429とバンの監視

監視なしのタスクキューはブラックボックスです: なぜ一部のタスクが失敗し、どのIPが疑わしいかを理解するのが難しいです。追跡すべき最小限のメトリックは次のとおりです:

各プロキシごとの429および403コードのレスポンス率
平均応答時間 — 遅延の増加はしばしばバンの前触れです
タスクに対する再試行の回数
リアルタイムのタスクキューの長さ — 増加している場合、ワーカーが処理できていません

セロリでは、これを監視するためにFlowerを使用するのが便利です — タスクとワーカーのリアルタイム監視用のウェブインターフェースです。BullMQには同様の機能を持つBull Boardがあります。RQにはRQ Dashboardがあります。これらの3つのツールは、タスクのステータス、失敗した試行の数を表示し、プールから一時的に除外すべきプロキシを迅速に特定できます。

良いプラクティスは、連続してエラー429が発生したIPのプロキシを「隔離中」として15〜30分間マークし、成功したテストリクエストの後にのみプールに戻すことです。これにより、バンされたIPがタスクを受け取り続け、ワーカーの時間を無駄にする状況を防ぎます。

結論

タスクキューはリクエストのスケーリングの問題を解決しますが、IPに対する制限から自動的に保護するわけではありません — セロリ、RQ、またはBullMQ内でのプロキシの適切なローテーションが必要です。ツールの選択はスタックに依存します: Pythonプロジェクトでは、プロダクション用にセロリを選び、簡単なタスクにはRQを選ぶことが多いです。Node.js開発者は、組み込みのレートリミッターを持つBullMQを選ぶことが一般的です。

選択したキューに関係なく、結果はプロキシプールの質に大きく依存します。迅速な大量スクレイピングが目的であれば、速度のためにデータセンターのプロキシを検討し、ソーシャルメディアやマーケットプレイスでの作業には、アンチボットシステムへの忠誠が重要なため、レジデンシャルまたはモバイルIPが適しています — これにより、エラー429やバンの割合が減少し、再試行の数が直接的に減少し、タスクキュー全体の実行が加速します。