到目前为止，索引中的文档是一次性灌入的。真实搜索引擎的内容不断变化——新页面发布、旧页面修改、过期页面删除。搜索引擎需要持续追踪这些变化，保持索引的新鲜度。

本篇实现持续采集系统：条件请求避免无谓下载，自适应调度控制抓取节奏，有界队列和背压防止资源耗尽，幂等写入保证重试安全，状态持久化保证重启不丢进度。

条件请求

HTTP 条件请求机制

HTTP 协议内置了条件请求机制，避免重复下载未变化的内容。

Last-Modified / If-Modified-Since：

record CrawlMetadata(
    String url,
    String etag,
    String lastModified,
    long lastCrawledAt,
    int consecutiveNotModified,
    int consecutive404
) {}

HttpResponse<String> conditionalFetch(CrawlMetadata meta, HttpClient client)
        throws IOException, InterruptedException {
    HttpRequest.Builder builder = HttpRequest.newBuilder()
        .uri(URI.create(meta.url()))
        .timeout(Duration.ofSeconds(10));

    if (meta.lastModified() != null) {
        builder.header("If-Modified-Since", meta.lastModified());
    }
    if (meta.etag() != null) {
        builder.header("If-None-Match", meta.etag());
    }

    return client.send(builder.build(), HttpResponse.BodyHandlers.ofString());
}

处理响应：

CrawlResult handleResponse(HttpResponse<String> response, CrawlMetadata meta) {
    return switch (response.statusCode()) {
        case 200 -> {
            String etag = response.headers().firstValue("ETag").orElse(null);
            String lastMod = response.headers()
                .firstValue("Last-Modified").orElse(null);
            yield new CrawlResult(
                CrawlStatus.UPDATED,
                response.body(),
                new CrawlMetadata(meta.url(), etag, lastMod,
                    System.currentTimeMillis(), 0, 0));
        }
        case 304 -> new CrawlResult(
            CrawlStatus.NOT_MODIFIED, null,
            new CrawlMetadata(meta.url(), meta.etag(), meta.lastModified(),
                System.currentTimeMillis(),
                meta.consecutiveNotModified() + 1, 0));
        case 404, 410 -> new CrawlResult(
            CrawlStatus.NOT_FOUND, null,
            new CrawlMetadata(meta.url(), null, null,
                System.currentTimeMillis(), 0,
                meta.consecutive404() + 1));
        default -> new CrawlResult(
            CrawlStatus.ERROR, null, meta);
    };
}

enum CrawlStatus { UPDATED, NOT_MODIFIED, NOT_FOUND, ERROR }
record CrawlResult(CrawlStatus status, String body, CrawlMetadata metadata) {}

304 响应没有 body，节省带宽和解析开销。对于大型 HTML 页面，这个差异很明显。

自适应重抓调度

变化频率驱动调度间隔

class AdaptiveScheduler {
    private static final Duration MIN_INTERVAL = Duration.ofMinutes(30);
    private static final Duration MAX_INTERVAL = Duration.ofDays(30);
    private static final double BACKOFF_FACTOR = 1.5;
    private static final double SPEEDUP_FACTOR = 0.5;

    Duration nextInterval(CrawlMetadata meta, Duration currentInterval) {
        if (meta.consecutive404() > 0) {
            // 404 也退避——不要频繁重试不存在的页面
            return clamp(currentInterval.multipliedBy(2));
        }
        if (meta.consecutiveNotModified() > 0) {
            // 未变化 → 降低抓取频率
            long newMillis = (long)(currentInterval.toMillis() * BACKOFF_FACTOR);
            return clamp(Duration.ofMillis(newMillis));
        }
        // 有变化 → 提高抓取频率
        long newMillis = (long)(currentInterval.toMillis() * SPEEDUP_FACTOR);
        return clamp(Duration.ofMillis(newMillis));
    }

    private Duration clamp(Duration d) {
        if (d.compareTo(MIN_INTERVAL) < 0) return MIN_INTERVAL;
        if (d.compareTo(MAX_INTERVAL) > 0) return MAX_INTERVAL;
        return d;
    }
}

连续多次 304 → 间隔越来越长。一旦检测到变化 → 间隔缩短。

URL 优先级队列

按 nextCrawlAt 排序，最急需抓取的 URL 排在前面：

record ScheduledUrl(String url, long nextCrawlAt, Duration interval)
        implements Comparable<ScheduledUrl> {
    @Override
    public int compareTo(ScheduledUrl other) {
        return Long.compare(this.nextCrawlAt, other.nextCrawlAt);
    }
}

PriorityQueue<ScheduledUrl> urlFrontier = new PriorityQueue<>();

有界队列与背压

有界工作队列

class BoundedCrawlQueue {
    private final ArrayBlockingQueue<String> queue;
    private final int capacity;

    BoundedCrawlQueue(int capacity) {
        this.capacity = capacity;
        this.queue = new ArrayBlockingQueue<>(capacity);
    }

    boolean offer(String url, Duration timeout)
            throws InterruptedException {
        return queue.offer(url, timeout.toMillis(), TimeUnit.MILLISECONDS);
    }

    String take() throws InterruptedException {
        return queue.take();
    }

    int size() { return queue.size(); }
    int remainingCapacity() { return queue.remainingCapacity(); }
}

capacity 设为 10000——超过这个数量说明消费速度跟不上发现速度。

背压信号

void submitUrl(String url, BoundedCrawlQueue queue) {
    boolean accepted = false;
    try {
        accepted = queue.offer(url, Duration.ofSeconds(5));
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }

    if (!accepted) {
        // 背压：队列满，记录到溢出日志
        spillToDisk(url);
    }
}

溢出到磁盘的 URL 在队列有空间时重新加载。

域名速率限制

class DomainRateLimiter {
    private final Map<String, Long> lastRequestTime = new ConcurrentHashMap<>();
    private final long minIntervalMillis;

    DomainRateLimiter(Duration minInterval) {
        this.minIntervalMillis = minInterval.toMillis();
    }

    void waitIfNeeded(String domain) throws InterruptedException {
        long now = System.currentTimeMillis();
        Long last = lastRequestTime.get(domain);
        if (last != null) {
            long elapsed = now - last;
            if (elapsed < minIntervalMillis) {
                Thread.sleep(minIntervalMillis - elapsed);
            }
        }
        lastRequestTime.put(domain, System.currentTimeMillis());
    }
}

默认间隔 1 秒/域名。遵守 robots.txt 中 Crawl-delay 指令的值（如果更大）。

幂等写入

问题

同一个 URL 可能因为重试、重复发现、调度重叠等原因被多次抓取。如果每次抓取都往索引里追加一个文档，同一页面会出现多个副本。

updateDocument

Lucene 的 updateDocument 是原子的"先删后加"操作：

void indexPage(String url, String title, String body,
               float[][] chunkVectors, IndexWriter writer)
        throws IOException {
    // 先按 URL 删除旧文档（如果存在），再添加新文档
    Term urlTerm = new Term("url", url);

    List<Document> docs = buildDocuments(url, title, body, chunkVectors);
    for (Document doc : docs) {
        writer.updateDocument(urlTerm, doc);
    }
}

相同 URL 无论写入多少次，索引中只保留最后一次的版本。

内容指纹去重

有时 URL 不同但内容相同（镜像、URL 参数差异）。用内容哈希去重：

String contentFingerprint(String body) {
    return Integer.toHexString(body.hashCode());
}

boolean shouldIndex(String url, String body, Set<String> seenFingerprints) {
    String fp = contentFingerprint(body);
    if (seenFingerprints.contains(fp)) {
        return false;  // 内容重复，跳过
    }
    seenFingerprints.add(fp);
    return true;
}

删除检测

软删除策略

单次 404 不立即删除——可能是服务器临时故障：

void handleNotFound(CrawlMetadata meta, IndexWriter writer)
        throws IOException {
    if (meta.consecutive404() >= 3) {
        // 连续 3 次 404，确认删除
        writer.deleteDocuments(new Term("url", meta.url()));
        removeCrawlMetadata(meta.url());
    }
    // < 3 次：保留文档，下次再检查
}

410 Gone 表示服务器明确告知页面永久删除——可以立即删除：

if (response.statusCode() == 410) {
    writer.deleteDocuments(new Term("url", url));
    removeCrawlMetadata(url);
}

状态持久化与重启恢复

持久化抓取状态

用 SQLite 存储每个 URL 的抓取元数据：

class CrawlStateStore {
    private final Connection conn;

    CrawlStateStore(Path dbPath) throws SQLException {
        conn = DriverManager.getConnection("jdbc:sqlite:" + dbPath);
        conn.createStatement().execute("""
            CREATE TABLE IF NOT EXISTS crawl_state (
                url TEXT PRIMARY KEY,
                etag TEXT,
                last_modified TEXT,
                last_crawled_at INTEGER,
                consecutive_not_modified INTEGER DEFAULT 0,
                consecutive_404 INTEGER DEFAULT 0,
                interval_millis INTEGER DEFAULT 3600000,
                next_crawl_at INTEGER
            )
        """);
    }

    void save(CrawlMetadata meta, Duration interval) throws SQLException {
        long nextCrawlAt = meta.lastCrawledAt() + interval.toMillis();
        try (PreparedStatement ps = conn.prepareStatement("""
                INSERT OR REPLACE INTO crawl_state
                (url, etag, last_modified, last_crawled_at,
                 consecutive_not_modified, consecutive_404,
                 interval_millis, next_crawl_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                """)) {
            ps.setString(1, meta.url());
            ps.setString(2, meta.etag());
            ps.setString(3, meta.lastModified());
            ps.setLong(4, meta.lastCrawledAt());
            ps.setInt(5, meta.consecutiveNotModified());
            ps.setInt(6, meta.consecutive404());
            ps.setLong(7, interval.toMillis());
            ps.setLong(8, nextCrawlAt);
            ps.executeUpdate();
        }
    }

    List<ScheduledUrl> loadDueUrls(long now) throws SQLException {
        try (PreparedStatement ps = conn.prepareStatement(
                "SELECT url, next_crawl_at, interval_millis FROM crawl_state WHERE next_crawl_at <= ?")) {
            ps.setLong(1, now);
            ResultSet rs = ps.executeQuery();
            List<ScheduledUrl> urls = new ArrayList<>();
            while (rs.next()) {
                urls.add(new ScheduledUrl(
                    rs.getString("url"),
                    rs.getLong("next_crawl_at"),
                    Duration.ofMillis(rs.getLong("interval_millis"))));
            }
            return urls;
        }
    }
}

重启恢复

void startup(CrawlStateStore store, BoundedCrawlQueue queue)
        throws SQLException, InterruptedException {
    long now = System.currentTimeMillis();
    List<ScheduledUrl> dueUrls = store.loadDueUrls(now);

    System.out.printf("恢复 %d 个到期 URL\n", dueUrls.size());

    for (ScheduledUrl url : dueUrls) {
        queue.offer(url.url(), Duration.ofSeconds(1));
    }
}

重启后从 SQLite 加载所有 nextCrawlAt <= now 的 URL，放入工作队列。不需要重新发现这些 URL。

完整采集循环

主循环

void crawlLoop(BoundedCrawlQueue queue, CrawlStateStore store,
               DomainRateLimiter limiter, AdaptiveScheduler scheduler,
               IndexWriter writer, HttpClient httpClient) {
    while (!Thread.currentThread().isInterrupted()) {
        try {
            String url = queue.take();
            String domain = URI.create(url).getHost();
            limiter.waitIfNeeded(domain);

            CrawlMetadata meta = store.load(url);
            HttpResponse<String> response = conditionalFetch(meta, httpClient);
            CrawlResult result = handleResponse(response, meta);

            switch (result.status()) {
                case UPDATED -> {
                    indexPage(url, extractTitle(result.body()),
                        result.body(), encodeChunks(result.body()), writer);
                    writer.commit();
                }
                case NOT_FOUND -> handleNotFound(result.metadata(), writer);
                case NOT_MODIFIED -> { /* 无操作 */ }
                case ERROR -> { /* 记录错误，稍后重试 */ }
            }

            Duration interval = scheduler.nextInterval(
                result.metadata(), meta != null ?
                    Duration.ofMillis(store.getInterval(url)) :
                    Duration.ofHours(1));
            store.save(result.metadata(), interval);

        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            break;
        } catch (Exception e) {
            // 记录错误，继续处理下一个 URL
        }
    }
}

URL 生命周期时间线

发现 URL
  ↓
首次抓取 → 200 OK → 索引写入 → 调度(interval=1h)
  ↓
重抓 → 304 Not Modified → interval×1.5 = 1.5h
  ↓
重抓 → 304 Not Modified → interval×1.5 = 2.25h
  ↓
重抓 → 200 OK(有变化) → 索引更新 → interval×0.5 = 1.125h
  ↓
重抓 → 404 → consecutive_404=1 → 继续调度
  ↓
重抓 → 404 → consecutive_404=2 → 继续调度
  ↓
重抓 → 404 → consecutive_404=3 → 从索引删除

── 服务重启 ──

恢复: 加载所有 nextCrawlAt <= now 的 URL → 放入队列 → 继续

当前局限

单线程采集——高吞吐场景需要多线程 + 线程池
内容指纹用 hashCode——生产环境应使用 SHA-256
SQLite 单写者——并发写入需要加锁或换用其他存储
没有 robots.txt 解析——需要增加 RobotsTxtParser

练习

实现条件请求，对比有条件请求和无条件请求的带宽消耗
用自适应调度器抓取 10 个 URL，观察间隔变化曲线
设置 capacity=100 的有界队列，验证背压行为
模拟抓取 → 重抓 → 404 → 删除的完整生命周期
杀掉进程后重启，验证抓取状态不丢失

从零构建现代搜索引擎（25）：新鲜度与持续采集