大文件处理

对于大型 JSON 文件（如日志、配置、数据导出），直接加载到内存可能导致内存溢出。json 库提供了多种高效的处理方式。

WARNING

ForeachFile 和 ForeachFileChunked 在迭代前会将整个文件加载到内存中。"分块"行为仅影响内存中数据的迭代方式，不影响文件的读取方式。对于真正需要控制内存的超大文件处理，请使用 NDJSONProcessor 配合 JSONL 格式，或使用 StreamIterator。

备选方案

方案	适用场景	内存占用
Processor.ForeachFile	结构化迭代处理文件	加载完整文件，逐条迭代
Processor.ForeachFileChunked	批量分块迭代处理	加载完整文件，分块迭代
NDJSONProcessor	逐行处理 JSONL 文件	内存可控，真正的流式处理

统一 API：Processor

配置选项

大文件处理配置已集成到 Config 中：

type Config struct {
    // ... 其他配置 ...

    // 大文件处理配置
    ChunkSize       int64 // 分块大小（默认 1MB）
    MaxMemory       int64 // 最大内存使用（默认 100MB）
    BufferSize      int   // 读取缓冲区大小（默认 64KB）
    SamplingEnabled bool  // 是否启用采样（默认 true）
    SampleSize      int   // 采样数量（默认 1000）
}

基本使用

package main

import (
    "log"
    "github.com/cybergodev/json"
)

func main() {
    // 创建 Processor（使用默认配置）
    processor, err := json.New()
    if err != nil {
        log.Fatal(err)
    }
    defer processor.Close()

    // 方式 1：逐条处理（推荐）
    count := 0
    err = processor.ForeachFile("large-data.json", func(key any, item *json.IterableValue) error {
        count++

        // 使用 IterableValue 便捷访问字段
        id := item.GetInt("id")
        name := item.GetString("name")
        email := item.GetString("email")

        // 支持路径访问嵌套属性
        city := item.GetString("profile.city")
        interests := item.GetArray("profile.interests")

        if count%10000 == 0 {
            log.Printf("已处理 %d 条记录，示例: id=%d name=%s email=%s city=%s 兴趣数=%d",
                count, id, name, email, city, len(interests))
        }
        return nil
    })

    if err != nil {
        log.Fatal(err)
    }
    log.Printf("处理完成，共 %d 条记录", count)
}

批量处理

// 方式 2：分批处理（适合批量写入数据库）
err := processor.ForeachFileChunked("large-data.json", 1000, func(chunk []*json.IterableValue) error {
    log.Printf("处理批次：%d 条记录", len(chunk))

    // 批量写入数据库
    for _, item := range chunk {
        id := item.GetInt("id")
        name := item.GetString("name")
        // ... 处理数据
    }
    return nil
})

带中断控制

// 方式 3：带中断控制（查找特定数据后停止）
// 返回 item.Break() 停止迭代，返回 nil 继续迭代
err := processor.ForeachFile("large-data.json", func(key any, item *json.IterableValue) error {
    id := item.GetInt("id")

    if id == targetID {
        // 找到目标，停止迭代
        fmt.Printf("找到目标: ID=%d, Name=%s\n", id, item.GetString("name"))
        return item.Break() // 停止迭代（返回中断信号）
    }

    return nil // 继续迭代
})

处理对象文件

// 方式 4：处理 JSON 对象文件（键值对结构）
// 文件格式：{"user1": {...}, "user2": {...}, ...}
err := processor.ForeachFile("config-map.json", func(key any, item *json.IterableValue) error {
    fmt.Printf("Key: %s, Name: %s\n", key, item.GetString("name"))
    return nil
})

自定义配置

// 自定义大文件处理配置
cfg := json.DefaultConfig()
cfg.ChunkSize = 10 * 1024 * 1024   // 10MB 分块
cfg.MaxMemory = 500 * 1024 * 1024  // 500MB 内存限制
cfg.BufferSize = 128 * 1024        // 128KB 缓冲区

processor, err := json.New(cfg)
if err != nil {
    panic(err)
}
defer processor.Close()

IterableValue 便捷方法

ForeachFile* 系列方法提供 IterableValue 接口，支持便捷的数据访问：

方法	说明	示例
Get(path)	获取值	item.Get("field")
GetString(path)	获取字符串	item.GetString("name")
GetInt(path)	获取整数	item.GetInt("id")
GetFloat64(path)	获取浮点数	item.GetFloat64("score")
GetBool(path)	获取布尔值	item.GetBool("active")
GetArray(path)	获取数组	item.GetArray("tags")
GetObject(path)	获取对象	item.GetObject("profile")
Exists(path)	检查字段是否存在	item.Exists("email")
IsNull(path)	检查是否为 null	item.IsNull("deleted_at")
IsEmpty(path)	检查是否为空	item.IsEmpty("notes")
Break()	返回中断信号	return item.Break()

支持路径导航

city := item.GetString("profile.address.city")      // 嵌套对象
firstTag := item.GetString("tags[0]")               // 数组索引
lastTag := item.GetString("tags[-1]")               // 负索引（最后一个）
nested := item.GetString("data.items[0].name")      // 复杂路径

流式处理配置

通过 Config 配置流式处理参数：

cfg := json.DefaultConfig()

// 大文件处理配置
cfg.ChunkSize = 10 * 1024 * 1024   // 10MB 分块
cfg.MaxMemory = 500 * 1024 * 1024  // 500MB 内存限制
cfg.BufferSize = 128 * 1024        // 128KB 缓冲区

processor, err := json.New(cfg)
if err != nil {
    panic(err)
}
defer processor.Close()

使用 StreamLinesInto 泛型函数

type User struct {
    Name string `json:"name"`
}

file, _ := os.Open("users.jsonl")
defer file.Close()

_, err := json.StreamLinesInto[User](file, func(lineNum int, user User) error {
    fmt.Printf("处理: %s\n", user.Name)
    return nil
})

并行处理

对于可并行处理的任务，可以使用多 goroutine：

package main

import (
    "sync"
    "github.com/cybergodev/json"
)

func main() {
    processor, err := json.New()
    if err != nil {
        panic(err)
    }
    defer processor.Close()

    // 使用 worker pool
    workers := 4
    items := make(chan any, 100)
    var wg sync.WaitGroup

    // 启动 workers
    for i := 0; i < workers; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for item := range items {
                // 处理 item（替换为你的业务逻辑）
                _ = item
            }
        }(i)
    }

    // 流式读取并分发
    processor.ForeachFile("large-data.json", func(key any, item *json.IterableValue) error {
        items <- item.Get("")
        return nil
    })

    close(items)
    wg.Wait()
}

性能优化建议

内存控制

// 根据可用内存配置
cfg := json.DefaultConfig()
cfg.MaxMemory = 500 * 1024 * 1024 // 500MB
cfg.ChunkSize = 10 * 1024 * 1024  // 10MB

processor, err := json.New(cfg)
if err != nil {
    panic(err)
}
defer processor.Close()

最佳实践

1. 预估文件大小：处理前检查文件大小，选择合适的策略
2. 设置内存限制：使用 MaxMemory 防止 OOM
3. 批量提交：积累一定数量后批量写入数据库
4. 错误处理：实现 JSONLContinueOnErr 或记录失败条目
5. 进度监控：定期输出处理进度

选择指南

文件大小	推荐方案	示例
< 10MB	直接加载	json.ParseAny + Get
10-100MB	Processor.ForeachFile	逐条处理
100MB-1GB	Processor.ForeachFileChunked	分块迭代处理
> 1GB	NDJSONProcessor / JSONL 格式	真正的流式处理，内存可控

API 参考

本节汇总大文件处理相关 API 的函数签名与参数表，便于快速查阅。

Processor 方法

ForeachFile

签名：func (p *Processor) ForeachFile(filePath string, fn func(key any, item *IterableValue) error, cfg ...Config) error

逐条处理大文件中的 JSON 数组元素。完整用法见基本使用与中断控制。

参数

名称	类型	说明
filePath	string	JSON 文件路径
fn	func(key any, item *IterableValue) error	处理回调

回调返回值

返回值	说明
nil	继续处理下一项
item.Break()	停止迭代，不返回错误
其他 error	停止迭代并返回错误

ForeachFileChunked

签名：func (p *Processor) ForeachFileChunked(filePath string, chunkSize int, fn func(chunk []*IterableValue) error, cfg ...Config) (err error)

分批处理大文件，每次处理指定数量的元素。用法见批量处理。

参数

名称	类型	说明
filePath	string	JSON 文件路径
chunkSize	int	每批元素数量
fn	func(chunk []*IterableValue) error	批处理回调

ForeachFileWithPath

签名：func (p *Processor) ForeachFileWithPath(filePath, path string, fn func(key any, item *IterableValue) error, cfg ...Config) error

处理文件中指定路径的 JSON 数组或对象。

参数

名称	类型	说明
filePath	string	JSON 文件路径
path	string	JSON 路径表达式
fn	func(key any, item *IterableValue) error	处理回调

// 处理文件中 users 数组的每个元素
err := p.ForeachFileWithPath("data.json", "users", func(key any, item *json.IterableValue) error {
    fmt.Printf("Name: %s\n", item.GetString("name"))
    return nil
})

ForeachFileNested

签名：func (p *Processor) ForeachFileNested(filePath string, fn func(key any, item *IterableValue) error, cfg ...Config) error

递归遍历文件中的所有嵌套 JSON 结构。

// 递归遍历所有嵌套元素
err := p.ForeachFileNested("data.json", func(key any, item *json.IterableValue) error {
    fmt.Printf("Key: %v, Type: %T\n", key, item.GetData())
    return nil
})

包级函数

除了 Processor 方法外，以下函数可以直接调用，无需创建 Processor 实例。它们内部使用全局处理器。

ForeachFile（包级函数）

签名：func ForeachFile(filePath string, fn func(key any, item *IterableValue) error, cfg ...Config) error

从文件加载 JSON 并迭代。

err := json.ForeachFile("data.json", func(key any, item *json.IterableValue) error {
    fmt.Printf("[%v] %v\n", key, item.GetData())
    return nil
})

ForeachFileWithPath（包级函数）

签名：func ForeachFileWithPath(filePath, path string, fn func(key any, item *IterableValue) error, cfg ...Config) error

从文件加载 JSON 并按路径迭代。

err := json.ForeachFileWithPath("data.json", "users", func(key any, item *json.IterableValue) error {
    name := item.GetString("name")
    fmt.Printf("用户: %s\n", name)
    return nil
})

ForeachFileChunked（包级函数）

签名：func ForeachFileChunked(filePath string, chunkSize int, fn func(chunk []*IterableValue) error, cfg ...Config) error

分块迭代文件中的 JSON 数组。

err := json.ForeachFileChunked("large_data.json", 100, func(chunk []*json.IterableValue) error {
    for _, item := range chunk {
        processItem(item)
    }
    return nil
})

ForeachFileNested（包级函数）

签名：func ForeachFileNested(filePath string, fn func(key any, item *IterableValue) error, cfg ...Config) error

从文件加载 JSON 并递归迭代所有嵌套结构。

err := json.ForeachFileNested("config.json", func(key any, item *json.IterableValue) error {
    fmt.Printf("路径: %v, 类型: %T\n", key, item.GetData())
    return nil
})

相关

• NDJSON 处理器 — JSONL/NDJSON 流式处理
• JSONLWriter — JSONL 写入器

下一步

• API 文档 — 完整 API 参考