Go Badger 怎么批量写入并处理 Flush 错误

Go Badger 的批量写入关键不在于把更多的 Set 塞进一个对象，而在于把“收集写入”和“提交写入”分开处理。通常用 db.NewWriteBatch() 创建批次，在批次内逐条调用 Set 或 SetEntry，最后只把 Flush() 当作提交点；每一次写入和最终刷新都要保留错误。

官方地址：https://github.com/dgraph-io/badger/

本文只讨论 Badger v4 的 WriteBatch、Flush、Cancel 和错误处理，不把它扩展成数据库选型或通用事务教程。

先分清 Set 错误和 Flush 错误

WriteBatch.Set 的错误表示当前批次在收集数据时已经遇到问题，例如输入不符合 Badger 的写入约束；Flush 的错误则表示提交阶段返回了错误，或者返回了批次之前已经记录的错误。官方 API 还说明，一旦批次已经记录错误，后续不会继续执行提交。

因此，下面两件事不能混为一谈：

• 某条记录调用 Set 失败：停止继续追加，记录这条输入的位置。
• 所有 Set 都成功但 Flush 失败：把整个批次视为未完成，保留批次边界，再按业务策略恢复。

用有限大小的 WriteBatch 收集写入

批量大小不要只按记录条数决定，还要考虑单条 value 的大小和失败后的恢复成本。可以先用固定数量作为业务层分块边界，再通过 SetMaxPendingTxns 调整待处理事务上限。该方法应在开始写入前调用，官方文档给出的默认值是 16。

下面的示例将一批输入转换为 Badger 写入。示例故意把批次的起始游标交给调用方保存，这样调用方可以在外层记录“哪一批已经完成”，而不是把恢复状态藏在数据库内部。

package batchwrite

import (
    "fmt"

    badger "github.com/dgraph-io/badger/v4"
)

type Item struct {
    Key   string
    Value []byte
}

// writeOneBatch 只负责收集并提交一个有限批次。
// 调用方应在 Flush 成功后再推进游标，避免把未完成批次误记为成功。
func writeOneBatch(db *badger.DB, items []Item, maxPending int) error {
    wb := db.NewWriteBatch()
    // 如果中途提前返回，Cancel 负责释放尚未提交的批次状态。
    defer wb.Cancel()

    if maxPending > 0 {
        // 该上限应在第一次 Set 前设置，避免批次已经开始后再改变策略。
        wb.SetMaxPendingTxns(maxPending)
    }

    for index, item := range items {
        if item.Key == "" {
            // 空键属于输入错误，不应继续把后续记录追加到同一批次。
            return fmt.Errorf("item %d has empty key", index)
        }
        if err := wb.Set([]byte(item.Key), item.Value); err != nil {
            // Set 错误要立即返回，外层可以记录当前批次和记录下标。
            return fmt.Errorf("set item %d (%s): %w", index, item.Key, err)
        }
    }

    // Flush 是批次提交点；只有它成功，外层才应推进批次游标。
    if err := wb.Flush(); err != nil {
        return fmt.Errorf("flush %d items: %w", len(items), err)
    }
    return nil
}

这个函数没有在内部自动重试。原因是 Flush 已经是提交边界，遇到错误时直接复制同一批次并重试，可能让调用方误以为所有记录都处于同一种状态。是否重试应由外层根据错误、幂等键和持久化游标决定。

把 Flush 当作批次提交点

使用 WriteBatch 时，最容易遗漏的是“创建了批次但既没有 Flush，也没有 Cancel”。官方文档明确提醒，这会让事务 oracle 无法清理行提交时间映射，长期运行可能造成内存持续增长。因此，defer wb.Cancel() 应紧跟在创建批次之后；正常路径仍然显式调用 Flush。

建议外层保存三个字段：

这样即使应用在循环中途退出，也能从最后一个未完成批次开始处理，而不是凭“已经调用过多少次 Set”猜测进度。

Flush 出错时怎么恢复

恢复策略的核心是让每条输入拥有稳定、可重复计算的业务键。例如把订单号、对象版本或导入文件中的唯一行号编码到 key 中，而不是每次重试都生成新的随机 key。这样即使外层需要重新处理一个失败批次，也不会因为随机键产生一组无法覆盖的重复数据。

type Checkpoint struct {
    Start int
    End   int
}

// writeWithCheckpoint 只在整个批次 Flush 成功后返回新的游标。
// checkpoint 的持久化应由调用方完成，并与业务状态更新采用一致的顺序。
func writeWithCheckpoint(db *badger.DB, all []Item, start, size int) (Checkpoint, error) {
    if start >= len(all) {
        // 没有剩余输入时，返回空范围而不是创建空批次。
        return Checkpoint{Start: start, End: start}, nil
    }

    end := start + size
    if end > len(all) {
        // 最后一批只覆盖剩余数据，不人为填充占位记录。
        end = len(all)
    }
    if err := writeOneBatch(db, all[start:end], 16); err != nil {
        // 出错时保留原 start，下一次由调用方按幂等键决定是否恢复。
        return Checkpoint{Start: start, End: start}, err
    }
    return Checkpoint{Start: start, End: end}, nil
}

如果应用需要把 Badger 写入和外部检查点放进同一个不可分割的业务动作，就不能只依靠这个内存结构体；应把检查点设计成可重复写入的记录，并明确“Badger 成功、检查点失败”或相反顺序下的补偿方案。本文的示例只保证批次边界清楚，不声称跨系统事务。

常见误区和一张速查表

• 只检查最后一次 Flush：前面的 Set 已经失败时，批次不会按预期提交。
• 把 Cancel 当成回滚：官方说明已提交的写入仍会继续生效，Cancel 主要用于清理未完成的批次状态。
• 无限增大单批数据：单次失败后的重试成本、内存占用和游标恢复范围都会变大。
• 用随机键重试：同一业务记录可能被写成多条，后续无法判断哪条是有效版本。

总结一下：Badger 批量写入可以从 NewWriteBatch 开始，但可靠性取决于三个边界——每次 Set 的错误、最终 Flush 的提交结果，以及异常路径上的 Cancel。把批次游标和幂等键放在调用方，才有条件在 Flush 出错后安全恢复。

相关问题

WriteBatch 和普通事务 Set 有什么区别？

WriteBatch 面向一组批量写入，调用方最后显式 Flush；普通事务则围绕一次事务生命周期组织读写。选择哪一种，取决于你要处理的是批量导入边界，还是需要在事务内组合读取与写入。

Flush 成功后还需要调用 Cancel 吗？

代码可以保留创建后的 defer Cancel 作为提前返回的兜底，但真正的提交动作仍由 Flush 完成。不要把 Cancel 当作成功确认，也不要用它替代 Flush。

可以在 Flush 失败后直接再调用一次 Flush 吗？

不要把无条件重复 Flush 当作通用恢复方案。先记录错误和批次边界，再依据错误类型、幂等键和外部检查点设计恢复动作。