Go CSV 导入中断后怎么用 InputOffset 记录续传位置

我在处理大 CSV 导入时，最怕的不是程序退出，而是重启后只能从文件头重新跑一遍。Go 的 encoding/csv.Reader 提供了一个很适合做文件级检查点的接口：InputOffset() 返回最近一次完整读取记录的结束位置，也就是下一条记录的开始位置。把这个字节偏移在“业务处理成功之后”保存下来，重启时对同一个文件 Seek 到该位置，就能继续读取。

关键顺序只有一句话：先读取并成功处理当前记录，再保存 InputOffset；恢复时先确认还是同一个文件，再从这个偏移创建新的 CSV Reader。

先分清 InputOffset 记录的是什么

InputOffset 不是行号，也不是已经处理的记录数，而是输入流中的字节偏移。官方文档把它定义为“最近读取行的结束处和下一行开始处”。因此它适合交给支持 io.Seeker 的文件做断点恢复，但不适合直接当作业务主键。

还有一个容易忽略的点：CSV 字段可以包含换行。Reader 会把完整的 CSV record 读完后再更新当前位置，所以不要自己按文本换行数估算偏移。偏移是字节位置，中文内容不会按“字符个数”计数。

把偏移和处理结果放在同一个检查点里

检查点至少要包含偏移和文件身份。示例把文件名、文件大小和偏移写入 JSON；真实项目还可以增加文件摘要、导入批次号等字段。这里的 processed 只是观测信息，不能代替业务幂等。

type Checkpoint struct {
    FilePath  string `json:"file_path"`
    FileSize  int64  `json:"file_size"`
    Offset    int64  `json:"offset"`
    Processed int64  `json:"processed"`
}

func saveCheckpoint(path string, cp Checkpoint) error {
    data, err := json.MarshalIndent(cp, "", "  ")
    if err != nil {
        return fmt.Errorf("编码检查点: %w", err)
    }

    tmp := path + ".tmp"
    // 先写临时文件，再替换正式文件，避免进程中断留下半份 JSON。
    if err := os.WriteFile(tmp, data, 0o644); err != nil {
        return fmt.Errorf("写临时检查点: %w", err)
    }
    if err := os.Rename(tmp, path); err != nil {
        _ = os.Remove(tmp) // 清理失败的临时文件，保留原检查点。
        return fmt.Errorf("替换检查点: %w", err)
    }
    return nil
}

保存动作要放在一条记录的业务处理成功之后。如果先保存偏移、后提交业务，进程刚好在两者之间退出，恢复时会跳过这条尚未提交的记录。反过来，如果业务已经提交但偏移还没落盘，恢复后可能再次处理这一条，所以业务写入仍然需要幂等或可重试设计。

重启时用 Seek 恢复读取

恢复过程分四件事：打开同一个输入文件，读取检查点，确认偏移处于文件范围内，然后从偏移创建新的 csv.Reader。不要复用已经读过的 Reader，也不要把偏移直接传给 strings.Reader 之类无法对应原始文件字节位置的对象。

func openFromCheckpoint(csvPath, checkpointPath string) (*os.File, *csv.Reader, error) {
    file, err := os.Open(csvPath)
    if err != nil {
        return nil, nil, fmt.Errorf("打开 CSV: %w", err)
    }

    var cp Checkpoint
    data, err := os.ReadFile(checkpointPath)
    if err != nil {
        _ = file.Close() // 恢复失败时及时释放文件描述符。
        return nil, nil, fmt.Errorf("读取检查点: %w", err)
    }
    if err := json.Unmarshal(data, &cp); err != nil {
        _ = file.Close()
        return nil, nil, fmt.Errorf("解析检查点: %w", err)
    }

    info, err := file.Stat()
    if err != nil {
        _ = file.Close()
        return nil, nil, fmt.Errorf("读取文件信息: %w", err)
    }
    // 文件变短或偏移为负数时，不能盲目 Seek；应转入人工恢复或重新导入。
    if cp.Offset  info.Size() {
        _ = file.Close()
        return nil, nil, fmt.Errorf("检查点偏移越界: %d", cp.Offset)
    }
    if _, err := file.Seek(cp.Offset, io.SeekStart); err != nil {
        _ = file.Close()
        return nil, nil, fmt.Errorf("定位 CSV 偏移: %w", err)
    }

    reader := csv.NewReader(file)
    return file, reader, nil
}

从文件中间开始读时，Reader 看到的是偏移之后的内容。如果检查点本身就是一条完整 record 的结束位置，那么下一次 Read 会从下一条 record 开始。为了让恢复条件可靠，保存检查点时要和输入文件绑定；文件被覆盖、截断或重新生成后，旧偏移不能继续复用。

失败时怎么判断该重试还是回滚

线上运行手册里，我通常按下面几类信号处理：

如果输入文件来自上游定时生成，建议把文件摘要或稳定的对象版本也写进检查点。只比较文件名和大小并不总是足够：同名文件可能被替换，而且相同大小也可能内容不同。

小结：InputOffset 负责位置，业务层负责一致性

这套方案适合“一个稳定文件被逐条导入”的场景：Read 读取完整记录，业务处理成功后调用 InputOffset，再用原子方式保存检查点；重启时校验文件身份，Seek 到偏移并新建 Reader。它解决的是文件读取位置，不会自动解决重复提交、业务去重或上游文件替换问题。

相关问题

• InputOffset 是行号吗？不是，它是输入流中的字节偏移。
• CSV 字段包含换行还能续传吗？可以，前提是检查点来自完整 record 读取之后，而不是自行统计文本行。
• 保存 offset 后还会重复处理吗？可能。业务提交成功与检查点落盘之间如果中断，重启会再次读到上一条，因此业务写入要支持幂等。