从零构建现代搜索引擎（27）：节点故障、备份与恢复

上一篇把索引拆成了多个分片。分片解决了容量问题，但带来了新问题：一个分片进程挂了，那个分片上的文档就搜不到了。

本篇用只读副本实现读取冗余，用健康检查发现故障节点，用快照备份保证数据可恢复。目标：停一个进程后查询仍然可用（降级），从快照恢复后数据损失可量化（RPO/RTO）。

只读副本

为什么要副本

单个分片只有一个进程时：

• 进程挂了 → 该分片所有文档不可搜索
• 升级或重启 → 搜索服务中断

副本让同一个分片的数据存在于多个进程上。主节点负责写入，副本负责查询。

基于已提交快照的同步

主节点 writer.commit() 后，副本从共享目录打开新的 reader：

class PrimaryNode {
    private final IndexWriter writer;
    private final Path indexDir;

    void indexAndCommit(Document doc) throws IOException {
        writer.addDocument(doc);
        writer.commit();
        // commit 后 segment 文件写入磁盘
        // 副本可以看到新文件
    }
}

class ReplicaNode {
    private volatile DirectoryReader reader;
    private final Path indexDir;

    void refresh() throws IOException {
        DirectoryReader newReader = DirectoryReader.openIfChanged(reader);
        if (newReader != null) {
            DirectoryReader oldReader = this.reader;
            this.reader = newReader;
            oldReader.close();
        }
    }

    TopDocs search(Query query, int topK) throws IOException {
        IndexSearcher searcher = new IndexSearcher(reader);
        return searcher.search(query, topK);
    }
}

DirectoryReader.openIfChanged() 检查目录中是否有新的 commit point。有新 commit 时返回新 reader，否则返回 null。

副本不接受写入

副本只打开 DirectoryReader，不创建 IndexWriter。好处：

• 不需要写锁——多个副本可以同时读同一个目录
• 不需要处理并发写入冲突
• 副本故障不会损坏索引

同步延迟

主节点 commit 到副本 refresh 之间有延迟：

• 共享文件系统（NFS/本地磁盘）：取决于文件系统缓存刷新，通常 < 1 秒
• 文件复制：取决于网络带宽和 segment 大小，通常秒级

副本可能返回稍旧的数据——最终一致性，不是强一致性。搜索场景通常可以接受。

健康检查

健康状态

enum NodeHealth { GREEN, YELLOW, RED }

record NodeStatus(
    int shardId,
    boolean isPrimary,
    NodeHealth health,
    long lastSuccessAt,
    long lastCheckAt,
    String errorMessage
) {}

健康检查器

class HealthChecker {
    private final Map<String, NodeStatus> nodeStatuses = new ConcurrentHashMap<>();
    private final Duration timeout;
    private final Duration checkInterval;

    HealthChecker(Duration timeout, Duration checkInterval) {
        this.timeout = timeout;
        this.checkInterval = checkInterval;
    }

    void checkNode(ShardClient client, int shardId, boolean isPrimary) {
        String nodeKey = shardId + (isPrimary ? "-primary" : "-replica");
        long now = System.currentTimeMillis();

        try {
            // 发送轻量查询作为健康探针
            ShardResult result = client.search("_health", 1);
            nodeStatuses.put(nodeKey, new NodeStatus(
                shardId, isPrimary, NodeHealth.GREEN, now, now, null));
        } catch (Exception e) {
            NodeStatus prev = nodeStatuses.get(nodeKey);
            long lastSuccess = prev != null ? prev.lastSuccessAt() : 0;

            NodeHealth health;
            if (now - lastSuccess > timeout.toMillis() * 3) {
                health = NodeHealth.RED;
            } else {
                health = NodeHealth.YELLOW;
            }

            nodeStatuses.put(nodeKey, new NodeStatus(
                shardId, isPrimary, health, lastSuccess, now,
                e.getMessage()));
        }
    }

    List<NodeStatus> getUnhealthyNodes() {
        return nodeStatuses.values().stream()
            .filter(s -> s.health() != NodeHealth.GREEN)
            .toList();
    }
}

健康检查每 5 秒运行一次。连续 3 次失败（15 秒）标记为 RED。

查询路由集成

ShardClient selectNode(int shardId) {
    ShardClient primary = getPrimary(shardId);
    List<ShardClient> replicas = getReplicas(shardId);

    // 优先用健康的副本（分担主节点负载）
    for (ShardClient replica : replicas) {
        if (isHealthy(replica)) return replica;
    }

    // 副本都不可用，用主节点
    if (isHealthy(primary)) return primary;

    // 主节点也不可用
    return null;  // 该分片不可用
}

部分失败处理

降级查询

SearchResult searchWithDegradation(String queryText, int topK) {
    List<CompletableFuture<ShardResult>> futures = new ArrayList<>();
    List<Integer> unavailableShards = new ArrayList<>();

    for (int i = 0; i < numShards; i++) {
        ShardClient client = selectNode(i);
        if (client == null) {
            unavailableShards.add(i);
            continue;
        }
        final ShardClient c = client;
        futures.add(CompletableFuture.supplyAsync(
            () -> c.search(queryText, topK)));
    }

    List<ShardResult> results = futures.stream()
        .map(f -> {
            try {
                return f.get(200, TimeUnit.MILLISECONDS);
            } catch (Exception e) {
                return ShardResult.EMPTY;
            }
        })
        .toList();

    SearchResult merged = mergeResults(results, topK);

    if (!unavailableShards.isEmpty()) {
        System.out.printf("降级查询：分片 %s 不可用，结果可能不完整\n",
            unavailableShards);
    }

    return new SearchResult(
        merged.hits(),
        !unavailableShards.isEmpty()  // partial
    );
}

响应头中标记 X-Partial-Results: true 和 X-Missing-Shards: 1,2，让客户端知道结果不完整。

快照备份

创建快照

class SnapshotManager {
    private final IndexWriter writer;
    private final SnapshotDeletionPolicy snapshotPolicy;

    SnapshotManager(IndexWriter writer) {
        this.writer = writer;
        this.snapshotPolicy = (SnapshotDeletionPolicy)
            writer.getConfig().getIndexDeletionPolicy();
    }

    Path createSnapshot(Path backupDir) throws IOException {
        writer.commit();
        IndexCommit commit = snapshotPolicy.snapshot();

        try {
            Path snapshotDir = backupDir.resolve(
                "snapshot-" + commit.getGeneration());
            Files.createDirectories(snapshotDir);

            // 复制 commit 包含的所有文件
            Directory sourceDir = commit.getDirectory();
            for (String fileName : commit.getFileNames()) {
                Path target = snapshotDir.resolve(fileName);
                Files.copy(
                    Path.of(sourceDir.toString(), fileName),
                    target);
            }

            return snapshotDir;
        } finally {
            snapshotPolicy.release(commit);
        }
    }
}

SnapshotDeletionPolicy 防止备份期间 Lucene 合并时删除正在复制的 segment 文件。

备份调度

void scheduleBackups(SnapshotManager manager, Path backupDir,
                     Duration interval) {
    Thread.ofVirtual().start(() -> {
        while (!Thread.currentThread().isInterrupted()) {
            try {
                Path snapshot = manager.createSnapshot(backupDir);
                System.out.printf("备份完成: %s\n", snapshot);

                // 清理旧备份，只保留最近 3 个
                cleanOldSnapshots(backupDir, 3);

                Thread.sleep(interval);
            } catch (InterruptedException e) {
                break;
            } catch (Exception e) {
                System.err.printf("备份失败: %s\n", e.getMessage());
            }
        }
    });
}

从快照恢复

恢复流程

class RecoveryManager {

    void recover(Path snapshotDir, Path indexDir, Analyzer analyzer)
            throws IOException {
        long startTime = System.currentTimeMillis();

        // 1. 清空当前索引目录
        deleteDirectoryContents(indexDir);

        // 2. 复制快照文件到索引目录
        for (Path file : Files.list(snapshotDir).toList()) {
            Files.copy(file, indexDir.resolve(file.getFileName()));
        }

        // 3. 验证索引完整性
        try (DirectoryReader reader = DirectoryReader.open(
                FSDirectory.open(indexDir))) {
            int numDocs = reader.numDocs();
            System.out.printf("索引恢复完成: %d 文档\n", numDocs);
        }

        long rtoMillis = System.currentTimeMillis() - startTime;
        System.out.printf("RTO: %d ms\n", rtoMillis);
    }
}

测量 RPO 和 RTO

void measureRpoRto(IndexWriter writer, Path backupDir, Path indexDir)
        throws Exception {
    // 写入 1000 文档并提交
    for (int i = 0; i < 1000; i++) {
        writer.addDocument(createDoc(i));
    }
    writer.commit();
    long commitTime = System.currentTimeMillis();

    // 创建快照
    Path snapshot = snapshotManager.createSnapshot(backupDir);

    // 继续写入 50 文档（未提交）
    for (int i = 1000; i < 1050; i++) {
        writer.addDocument(createDoc(i));
    }
    long crashTime = System.currentTimeMillis();

    // 模拟故障
    writer.close();

    // 恢复
    long recoveryStart = System.currentTimeMillis();
    recoveryManager.recover(snapshot, indexDir, analyzer);
    long recoveryEnd = System.currentTimeMillis();

    // 检查文档数
    try (DirectoryReader reader = DirectoryReader.open(
            FSDirectory.open(indexDir))) {
        int recovered = reader.numDocs();
        int lost = 1050 - recovered;

        System.out.printf("""
            === RPO/RTO 报告 ===
            最后提交: %d 文档
            故障时: %d 文档
            恢复后: %d 文档
            丢失: %d 文档
            RPO (数据): %d 文档 (%d ms 的写入)
            RTO (时间): %d ms
            """,
            1000, 1050, recovered, lost,
            lost, crashTime - commitTime,
            recoveryEnd - recoveryStart);
    }
}

预期输出：

=== RPO/RTO 报告 ===
最后提交: 1000 文档
故障时: 1050 文档
恢复后: 1000 文档
丢失: 50 文档
RPO (数据): 50 文档 (320 ms 的写入)
RTO (时间): 1250 ms

RPO = 最后 commit 到故障之间的写入。减小 RPO 的方法：更频繁地 commit。代价：commit 有 I/O 开销。

降级演示

停一个进程

初始状态: 3 分片 × (1 主 + 1 副本) = 6 个进程

1. 查询 "搜索引擎" → 返回 10 条结果，partial=false
2. 停止 Shard1 的主节点
3. 健康检查检测到 Shard1 主节点 RED
4. 查询路由将 Shard1 的查询发送到副本
5. 查询 "搜索引擎" → 返回 10 条结果，partial=false

6. 停止 Shard1 的副本
7. 查询路由发现 Shard1 无可用节点
8. 查询 "搜索引擎" → 返回结果，partial=true，X-Missing-Shards: 1
9. 结果中缺少 Shard1 上的文档，但仍然返回其他分片的结果

当前局限

• 共享文件系统要求主节点和副本在同一台机器或 NFS 上——跨机器需要文件复制方案
• 没有自动故障转移——副本不会自动提升为主节点
• 没有共识协议——主节点故障后需要人工干预
• 备份是全量复制——增量备份需要额外实现

练习

1. 为一个分片配置一个只读副本，验证副本能读到主节点写入的数据
2. 实现健康检查，模拟节点故障后的状态转换
3. 停止一个分片的所有节点，验证降级查询行为
4. 创建快照备份并恢复，测量 RPO 和 RTO
5. 在不同 commit 频率下测量 RPO，画出 commit 频率 vs RPO 的关系

延伸阅读

• Elasticsearch: Index Recovery and Shard Allocation
• Apache Lucene: SnapshotDeletionPolicy, ReplicaNode
• Martin Kleppmann, “Designing Data-Intensive Applications”, Chapter 5: Replication