asyncio: как ограничить число одновременных соединений

Написал сборщик на aiohttp, кинул пять тысяч ссылок в asyncio.gather и получил гору таймаутов плюс подвисший процесс. Понял, что gather стартует всё разом.

Обернул в семафор:

sem = asyncio.Semaphore(50)

async def fetch(session, url): async with sem: async with session.get(url) as r: return await r.text() ```

Отказов стало меньше, но память на старте всё равно улетает в потолок. Чего я не понимаю?

Семафор придерживает только тех, кто уже дошёл до async with. Корутины ты создал все пять тысяч сразу, они висят в памяти и ждут очереди. Отсюда и потолок.

Ограничивать надо на входе в работу, там же, где ты подаёшь ссылки.

Классическая схема под это: очередь плюс фиксированное число рабочих. Тогда в памяти живёт ровно столько задач, сколько ты разрешил, остальное спокойно лежит строками в файле.

import asyncio, aiohttp

async def rabochiy(q, session, out): while True: url = await q.get() try: async with session.get(url, timeout=aiohttp.ClientTimeout(total=30, sock_read=15)) as r: out.append((url, r.status, len(await r.read()))) except Exception as e: out.append((url, 0, repr(e))) finally: q.task_done()

async def main(ssylki, n=50): q = asyncio.Queue(maxsize=n * 2) out = [] conn = aiohttp.TCPConnector(limit=n, limit_per_host=8, ttl_dns_cache=300) async with aiohttp.ClientSession(connector=conn) as session: rab = [asyncio.create_task(rabochiy(q, session, out)) for _ in range(n)] for u in ssylki: await q.put(u) await q.join() for t in rab: t.cancel() return out ```

maxsize у очереди тут важнее семафора: await q.put(u) сам притормозит подачу, когда рабочие перестают успевать. Это и есть обратное давление, после него можно лить хоть миллион строк построчно из файла.

Вот этого куска мне и не хватало. Переписал за вечер, память держится ровной полкой около 300 мегабайт, роста до упора больше нет.

а limit_per_host зачем отдельно, если общий лимит уже задан?

Настя, общий лимит делится между всеми узлами. Когда в списке один домен на пять тысяч ссылок, все 50 соединений уйдут туда, и сайт закроется от такого напора. limit_per_host держит вежливый потолок на каждый узел отдельно.

Второе, про таймаут. total считает всё вместе с чтением тела, поэтому на тяжёлых страницах его легко упереть. Ставь рядом sock_read, тогда медленное чтение оборвётся само, а нормальная большая страница доедет.

а если сервер отвечает 429, семафор тут вообще ни при чём?

Верно, 429 ловлю отдельно: поднимаю паузу и снижаю подачу на лету. Пока сделал грубо, общей переменной:

if r.status == 429:
    zaderzhka = min(zaderzhka * 2, 60)
    await asyncio.sleep(zaderzhka)

К числу рабочих привязан ещё выходной адрес, про это в ветке пока не сказали. Пока сборщик идёт с одного адреса, любой твой внутренний потолок упрётся в частоту на стороне сайта, и хоть сто корутин запускай.

Под такие прогоны я беру приватные адреса для сбора данных: ротация внутри пула идёт сама, трафик безлимитный, лимит потоков на стандартном пакете тысяча, так что в 50 рабочих упереться негде. Пул около 12 тысяч адресов, частота на каждый выход получается смешная, и сайт перестаёт видеть в тебе один настырный источник.

Дальше уже честно смотришь на отклик и подбираешь число рабочих под него.

Замечу про мелочь из примера выше: ttl_dns_cache=300 экономит прилично. Без него каждое новое соединение идёт в резолвер, и на тысячах адресов это заметный кусок времени.

спасибо, я всё это время лепила gather и удивлялась

Собрал у себя финальный набор, вдруг кому пригодится:

• очередь с maxsize, подача притормаживается сама;
• фиксированные рабочие, число подбираю по отклику;
• limit_per_host от 4 до 10;
• total и sock_read в таймауте оба;
• на 429 удвоение паузы с потолком в минуту.

Пять тысяч ссылок проходят за 11 минут, отказов единицы. Всем спасибо, тема закрыта.