Канал по Машинному обучению
@ai_machinelearning_big_data
Перейти
Разрываем интернет на Rust: свой многопоточный веб-краулер за вечер
Краулер это один из тех проектов, где Rust показывает свою истинную мощь. Тысячи одновременных соединений, разбор HTML, работа с очередями и разделяемым состоянием, жесткие требования по памяти. На Python вы быстро упрётесь в GIL, на Go получите хорошую производительность, но на Rust с tokio вы выжимаете из одной машины всё возможное. Сегодня построим краулер, который обходит сайт в несколько потоков, уважает robots.txt, ограничивает глубину, дедуплицирует ссылки, извлекает текст и заголовки страниц и складывает всё в JSONL файл. Это не игрушка. С небольшими допиливаниями вы сможете пихать в него миллионы URL.
Что именно мы строим
Разберёмся, что именно нам нужно от краулера. На вход он получает стартовый URL, максимальную глубину обхода и число параллельных рабочих. Дальше он качает страницы, вытаскивает ссылки из тегов a, фильтрует их по домену, чтобы не убежать на весь интернет, добавляет новые URL в очередь и записывает результат в файл. Главная идея схемы: один поставщик задач, много рабочих, один писатель результатов, и всё это связано каналами mpsc из tokio. Блокирующих вызовов нет, разделяемых мьютексов минимум.
Для всех кто хочет погрузится в Rust, много полезного в тг канале Rust code для разработчиков!
Зависимости
Создаём проект командой cargo new ferris-crawler и открываем Cargo.toml. Нам из всех крейтов нужно пять лошадок: tokio для асинхронного рантайма, reqwest для HTTP клиента, scraper для разбора HTML через CSS селекторы, url для нормализации адресов, serde и serde_json для вывода. Полный Cargo.toml выглядит вот так:
[package]name = "ferris-crawler"
version = "0.1.0"
edition = "2021"
[dependencies]tokio = { version = "1", features = ["full"] }
reqwest = { version = "0.12", features = ["gzip", "brotli", "rustls-tls"], default-features = false }
scraper = "0.20"
url = "2"
serde = { version = "1", features = ["derive"] }
serde_json = "1"
anyhow = "1"Сразу стоит прояснить пару моментов. Мы явно отключаем дефолтные фичи reqwest и включаем rustls-tls, чтобы не тащить за собой OpenSSL. Крейты работают на чистом Rust, сборка летит быстрее, а на слабых машинах это разница между 30 секундами и тремя минутами времени сборки. tokio берём с фичей full ради простоты, в проде выберете только нужные модули.
Модель данных и очередь
Перед кодом подумаем о типах. Задача в очереди это не просто строка URL, а структура с адресом и текущей глубиной обхода. Результат это структура с URL, заголовком страницы, длиной текста и временем ответа. Разделяемое состояние это множество уже посещённых ссылок и разрешённый домен. Будем использовать Arc и Mutex только там, где без этого никак, иначе латентность ползёт вверх.
use serde::Serialize;
use std::collections::HashSet;
use std::sync::Arc;
use tokio::sync::Mutex;
#[derive(Debug, Clone)]struct Job {
url: String,
depth: usize,
}
#[derive(Debug, Serialize)]struct PageResult {
url: String,
title: String,
text_len: usize,
elapsed_ms: u128,
status: u16,
}
#[derive(Clone)]struct Shared {
visited: Arc<Mutex<HashSet<String>>>,
host: String,
max_depth: usize,
}Обратите внимание на tokio::sync::Mutex вместо std::sync::Mutex. В асинхронном коде синхронный мьютекс опасен тем, что блокирует весь рабочий поток рантайма. Если всё же хотите синхронный, он быстрее при коротких локах, держите lock миллисекунды и ни в коем случае не вызывайте await под локом, иначе получите deadlock или очень странное поведение.
Рабочий, который делает всю работу
Сердце краулера это функция, которая берёт задачу из канала, скачивает страницу, парсит её и пихает обратно в канал новые URL. Каждый рабочий живёт в отдельной tokio task. Используем канал mpsc из tokio. Он синглрисиверский, поэтому оборачиваем Receiver в Arc<Mutex>, чтобы рабочие могли брать задачи по очереди. Если хочется идеала, берите async-channel, он MPMC без мьютекса.
use reqwest::Client;
use scraper::{Html, Selector};
use tokio::sync::mpsc::{Sender, Receiver};
use url::Url;
use std::time::Instant;
async fn worker(
id: usize,
rx: Arc<Mutex<Receiver<Job>>>,
tx: Sender<Job>,
out: Sender<PageResult>,
client: Client,
shared: Shared,
) {
loop {
let job = {
let mut guard = rx.lock().await;
guard.recv().await
};
let Some(job) = job else { break };
if job.depth > shared.max_depth { continue; }
// дедупликация: проверяем и сразу вставляем в одном локе
{
let mut v = shared.visited.lock().await;
if !v.insert(job.url.clone()) { continue; }
}
let started = Instant::now();
let resp = match client.get(&job.url).send().await {
Ok(r) => r,
Err(_) => continue,
};
let status = resp.status().as_u16();
let body = match resp.text().await {
Ok(b) => b,
Err(_) => continue,
};
let elapsed = started.elapsed().as_millis();
let (title, links, text_len) = parse_page(&body, &job.url);
let _ = out.send(PageResult {
url: job.url.clone(), title, text_len, elapsed_ms: elapsed, status,
}).await;
if job.depth < shared.max_depth {
for link in links {
if same_host(&link, &shared.host) {
let _ = tx.send(Job { url: link, depth: job.depth + 1 }).await;
}
}
}
let _ = id; // используйте для логов
}
}Обратите внимание на локальный блок вокруг recv. Мы освобождаем лок Receiver сразу после получения задачи, иначе все остальные рабочие будут ждать, пока первый обработает свой HTTP-запрос. Это классическая ошибка начинающих в tokio, она вроде бы не видна в коде, но ровняет всю производительность в однопоточный режим. И ещё let-else это один из самых красивых синтаксисов Rust 1.65+, пользуйтесь.
Парсинг HTML и нормализация URL
Крейт scraper работает на базе html5ever от Servo и поддерживает полный набор CSS-селекторов. Нам нужны всего два: title для заголовка и a[href] для ссылок. Большой плюс в том, что селекторы можно компилировать один раз и переиспользовать. Нормализация URL это отдельная история. Ссылки на страницах бывают относительными, с якорями, с mailto: и всяким мусором. Крейт url решает это при помощи метода join.
fn parse_page(html: &str, base: &str) -> (String, Vec<String>, usize) {
let doc = Html::parse_document(html);
let title_sel = Selector::parse("title").unwrap();
let a_sel = Selector::parse("a[href]").unwrap();
let title = doc
.select(&title_sel)
.next()
.map(|t| t.text().collect::<String>())
.unwrap_or_default()
.trim()
.to_string();
let base_url = Url::parse(base).ok();
let mut links = Vec::new();
if let Some(b) = base_url {
for a in doc.select(&a_sel) {
if let Some(href) = a.value().attr("href") {
if let Ok(joined) = b.join(href) {
let mut u = joined;
u.set_fragment(None); // режем якоря
if u.scheme() == "http" || u.scheme() == "https" {
links.push(u.to_string());
}
}
}
}
}
let text_len: usize = doc.root_element().text().map(|s| s.len()).sum();
(title, links, text_len)
}
fn same_host(url: &str, host: &str) -> bool {
Url::parse(url)
.ok()
.and_then(|u| u.host_str().map(|h| h.to_string()))
.map(|h| h == host)
.unwrap_or(false)
}Здесь обратите внимание на set_fragment(None). Без этого краулер будет считать, что example.com/page и example.com/page#top разные страницы, и вы потратите время на скачивание одного и того же содержимого десятки раз. Также фильтруем схемы: нам нужны только http и https, всё остальное игнорируем.
Собираем всё в main
Теперь свяжем всё воедино. В main открываем два канала: один для задач, второй для результатов. Запускаем N рабочих и одну задачу-писателя, которая берёт PageResult из канала и пишет строку JSON в файл. Режем формат JSONL выбран нарочно: его легко пихать в ClickHouse или в BigQuery, он стримится прямо в процессе работы и не требует хранить весь результат в памяти.
use tokio::fs::File;
use tokio::io::{AsyncWriteExt, BufWriter};
use tokio::sync::mpsc;
use anyhow::Result;
#[tokio::main]async fn main() -> Result<()> {
let start_url = std::env::args().nth(1).expect("укажите URL");
let max_depth: usize = std::env::var("DEPTH").ok().and_then(|s| s.parse().ok()).unwrap_or(2);
let workers: usize = std::env::var("WORKERS").ok().and_then(|s| s.parse().ok()).unwrap_or(16);
let host = Url::parse(&start_url)?.host_str().unwrap().to_string();
let shared = Shared {
visited: Arc::new(Mutex::new(HashSet::new())),
host: host.clone(),
max_depth,
};
let client = Client::builder()
.user_agent("FerrisCrawler/0.1")
.timeout(std::time::Duration::from_secs(10))
.build()?;
let (job_tx, job_rx) = mpsc::channel::<Job>(10_000);
let (out_tx, mut out_rx) = mpsc::channel::<PageResult>(10_000);
let job_rx = Arc::new(Mutex::new(job_rx));
job_tx.send(Job { url: start_url, depth: 0 }).await?;
// писатель результатов
let writer_task = tokio::spawn(async move {
let file = File::create("out.jsonl").await.unwrap();
let mut w = BufWriter::new(file);
while let Some(r) = out_rx.recv().await {
let line = serde_json::to_string(&r).unwrap();
let _ = w.write_all(line.as_bytes()).await;
let _ = w.write_all(b"\n").await;
}
let _ = w.flush().await;
});
// пул рабочих
let mut handles = Vec::new();
for i in 0..workers {
let h = tokio::spawn(worker(
i,
job_rx.clone(),
job_tx.clone(),
out_tx.clone(),
client.clone(),
shared.clone(),
));
handles.push(h);
}
drop(job_tx);
drop(out_tx);
for h in handles { let _ = h.await; }
let _ = writer_task.await;
Ok(())
}Самый хитрый момент во всём этом это drop(job_tx). Канал mpsc закрывается, когда умирают все отправители. Если мы будем держать оригинальный job_tx в main, рабочие никогда не выйдут из recv и программа повиснет навечно. Именно эта ошибка встречается в почти каждом любительском краулере на Rust. Решение в лоб это drop, более элегантное это использовать счётчик активных задач (AtomicUsize) и останавливать обход, когда все рабочие простаивают. Для базовой версии хватит и drop, в проде вам нужен более умный механизм завершения.
Запуск и результаты
Собираем в релизе и запускаем. Сборка займёт около минуты на современном MacBook, в релизе выходит бинарник от 8 до 10 МБ без зависимостей. Берём свой любимый блог или документацию какого-нибудь фреймворка:
$ cargo run --release -- https://docs.rs/tokio
$ DEPTH=3 WORKERS=32 cargo run --release -- https://example.com
$ wc -l out.jsonl
$ head -n 1 out.jsonl | jqНа свежем MacBook Air M1 этот краулер легко выдаёт от 600 до 1000 страниц в секунду на локальной копии сайта. На реальных ресурсах будет меньше из-за сетевых задержек, важный момент в том, что всё CPU-время уходит на парсинг HTML, а в сети вы сидите в await и никаких ресурсов не жрёте.
Практике на основе материалось кусра со Stepik – «Rust: полный курс разработчика. С нуля до профи»
Что добавить, чтобы это был реальный инструмент
База работает, но в прод так отдавать рано. Прежде всего добавьте разбор robots.txt: возьмите крейт robotstxt-rs или напишите свой парсер в районе 50 строк, это этика. Второе это превежливый лимит RPS на хост через крейт governor или простые tokio::time::sleep между запросами. Третье это персистентная очередь на SQLite или Redis, иначе при падении вы потеряете весь прогресс. Четвёртое это обработка редиректов и ETag, чтобы не качать страницы повторно при резапуске. Пятое это метрики в Prometheus через metrics + metrics-exporter-prometheus, без них вы никак не поймёте, почему краулер вдруг провисает.
Почему это получилось таким компактным
Весь краулер умещается в 200 с лишним строк Rust. Для сравнения, аналог на Python с aiohttp и BeautifulSoup входит в такое же число строк, но работает в несколько раз медленнее и жрёт память беспардонно на больших объёмах. Секрет в том, что Rust принуждает вас думать о владении и времени жизни прямо в момент написания. Вы вынуждены решать, где Arc, где клон, где ссылочный параметр. Именно это заставляет вас писать код, который не ляжет при росте нагрузки.
И ещё: обязательно прогоняйте код через cargo clippy и cargo fmt перед коммитом. Clippy для Rust тоже самое, что хороший линтер для любого другого языка, только умнее: он подскажет вам идиоматичные решения, поимает лишние клоны и расскажет, где вы зря взяли String вместо &str. Работать без него в Rust всё равно что варить борщ без свёклы.
+1
1
+1
1
+1
+1
+1
