并发与并行处理

CyberGo JSON 的所有操作都是并发安全的，并提供开箱即用的并行处理 API（ParallelIterator、并行 JSONL 流）。本页文档化线程安全语义、内置并行 API 与并发使用模式。

与性能优化页的分工

性能优化 的「并发处理」一节展示的是通用 Go 模式（sync.WaitGroup + 信号量 + Worker Pool）手动并发处理数组；本页文档化的是库内置并行 API，二者互补。

线程安全保证

Processor 是线程安全的主处理引擎（源码注释：Processor is the main JSON processing engine with thread safety）：

• 单个 Processor 实例可被多个 goroutine 共享——所有公开方法（Get/Set/Delete/Marshal 等）内部已用原子操作与并发治理（beginGovernedOp/endGovernedOp）保护。
• 包级函数（json.Get、json.GetString 等）共享一个全局 Processor，天然并发安全。
• PreParse 返回的 *ParsedJSON 可被并发读取——多个 goroutine 可同时对同一 ParsedJSON 调用 GetFromParsed。

何时不要共享

Processor 可共享，但不要跨 goroutine 共享可变的 Go 容器（如把 Get 返回的 map[string]any 交给多个 goroutine 改写）。库返回的容器默认是拷贝（除非开启 CacheSharedResults），改写返回值不影响缓存，但多 goroutine 改写同一容器仍需调用方自行加锁。

ParallelIterator 并行迭代器

ParallelIterator 利用多核 CPU 并行处理数组，内置 worker 池、错误聚合与 panic 恢复，比手写 goroutine 池更安全。

基础并行遍历

package main

import (
	"fmt"
	"sync"

	"github.com/cybergodev/json"
)

func main() {
	data := `{"items":[1,2,3,4,5,6,7,8]}`
	items := json.GetArray(data, "items")

	// worker 数默认取 Config.MaxConcurrency（受数组长度裁剪）
	iter := json.NewParallelIterator(items)
	defer iter.Close()

	var mu sync.Mutex
	var sum int64
	err := iter.ForEach(func(_ int, val any) error {
		mu.Lock()
		sum += int64(val.(float64))
		mu.Unlock()
		return nil
	})
	if err != nil {
		panic(err)
	}
	fmt.Printf("总和 = %d\n", sum)
	// 输出：总和 = 36
}

并行映射 Map

Map 并行变换每个元素，结果保持原始顺序（每个 worker 写入自己的索引位，无需加锁）。

package main

import (
	"fmt"

	"github.com/cybergodev/json"
)

func main() {
	data := `{"items":[1,2,3,4]}`
	items := json.GetArray(data, "items")

	iter := json.NewParallelIterator(items)
	defer iter.Close()

	// 并行映射：每个元素 *10，结果顺序与输入一致
	doubled, err := iter.Map(func(_ int, val any) (any, error) {
		return int(val.(float64)) * 10, nil
	})
	if err != nil {
		panic(err)
	}
	fmt.Println(doubled)
	// 输出：[10 20 30 40]
}

ParallelIterator API 一览

API	签名	说明
NewParallelIterator	func NewParallelIterator(data []any, cfg ...Config) *ParallelIterator	创建迭代器；worker 数取 cfg.MaxConcurrency
ForEach	func (it *ParallelIterator) ForEach(fn func(int, any) error) error	并行遍历，返回首个错误
ForEachWithContext	func (it *ParallelIterator) ForEachWithContext(ctx context.Context, fn func(int, any) error) error	支持 context 取消
ForEachBatch	func (it *ParallelIterator) ForEachBatch(batchSize int, fn func(int, []any) error) error	按批并行处理
Map	func (it *ParallelIterator) Map(transform func(int, any) (any, error)) ([]any, error)	并行变换，保序返回
Filter	func (it *ParallelIterator) Filter(predicate func(int, any) bool) []any	并行过滤
Close	func (it *ParallelIterator) Close()	释放资源（用完即调）

完整签名与用法见 迭代器类型。

错误与 panic 处理

ForEach 返回首个错误即停止派发新任务；worker 内的 panic 会被恢复（recover）并转为错误返回，不会击溃进程。需可取消时用 ForEachWithContext，在 ctx.Done() 时优雅退出。

并行 JSONL 流处理

处理大型 JSONL（NDJSON）文件时，StreamJSONLParallel 用多 worker 并行处理每一行。

package main

import (
	"fmt"
	"strings"
	"sync"

	"github.com/cybergodev/json"
)

func main() {
	// 模拟 JSONL 数据（每行一个 JSON 对象）
	jsonlData := `{"id":1,"score":95}
{"id":2,"score":82}
{"id":3,"score":78}
{"id":4,"score":90}`

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

	var mu sync.Mutex
	var total int64
	var count int64

	// 4 个 worker 并行处理每一行
	err = processor.StreamJSONLParallel(strings.NewReader(jsonlData), 4, func(lineNum int, item *json.IterableValue) error {
		score := int64(item.GetInt("score"))
		mu.Lock()
		total += score
		count++
		mu.Unlock()
		return nil
	})
	if err != nil {
		panic(err)
	}
	fmt.Printf("处理 %d 条，总分 %d\n", count, total)
	// 输出：处理 4 条，总分 345
}

API	说明
StreamJSONLParallel(reader, workers, fn)	多 worker 并行处理 JSONL
StreamJSONLParallelWithContext(ctx, reader, workers, fn)	同上，支持 context 取消与超时
StreamJSONLChunked(reader, chunkSize, fn)	按块处理，内存友好

完整签名与配置（JSONLWorkers/JSONLChunkSize 等）见 JSONL 处理 与 JSONL 流式。

行顺序

并行模式下回调的 lineNum 仍反映原始行号，但执行顺序不保证。若需保序输出，在回调中按 lineNum 写入预分配切片的对位索引。

全局处理器并发使用

SetGlobalProcessor 让所有包级函数共享同一个自定义 Processor，适合需要统一配置（缓存参数、钩子、安全限制）的多 goroutine 服务。

package main

import (
	"fmt"
	"sync"

	"github.com/cybergodev/json"
)

func main() {
	// 自定义全局处理器（所有包级函数共享，并发安全）
	cfg := json.DefaultConfig()
	processor, err := json.New(cfg)
	if err != nil {
		panic(err)
	}
	json.SetGlobalProcessor(processor) // 旧的全局 Processor 会被自动关闭
	defer json.ShutdownGlobalProcessor() // 应用退出时干净关闭

	data := `{"user":{"name":"Alice","age":30}}`

	// 多个 goroutine 并发使用包级函数（共享同一全局 Processor）
	var wg sync.WaitGroup
	results := make([]string, 3)
	for i := 0; i < 3; i++ {
		wg.Add(1)
		go func(idx int) {
			defer wg.Done()
			switch idx {
			case 0:
				results[idx] = json.GetString(data, "user.name")
			case 1:
				results[idx] = fmt.Sprintf("%d", json.GetInt(data, "user.age"))
			case 2:
				results[idx] = json.GetString(data, "user.name")
			}
		}(i)
	}
	wg.Wait()
	fmt.Println(results)
	// 输出：[Alice 30 Alice]
}

所有权转移

SetGlobalProcessor 后，该 Processor 的生命周期由全局管理——不要再手动 Close() 它，否则会与全局的关闭逻辑冲突。退出时调用 ShutdownGlobalProcessor() 即可干净关闭并释放资源。

并发限制 MaxConcurrency

Config.MaxConcurrency（默认 50）是单个 Processor 的软并发上限：用原子计数信号量限制在途操作数。达到上限时，新操作返回 ErrConcurrencyLimit（可重试）。

cfg := json.DefaultConfig()
cfg.MaxConcurrency = 100 // 提高单 Processor 并发上限

• ErrConcurrencyLimit 是可重试的暂时性错误（见 错误处理）。
• 并行流处理（StreamJSONLParallel）的 worker 数由参数显式指定，不与 MaxConcurrency 直接绑定，但共享同一治理槽位。
• ParallelIterator 的 worker 数取自 cfg.MaxConcurrency（默认 50），但会被数组长度裁剪。

最佳实践与陷阱

1. 复用 Processor，不要每请求新建

Processor 内部维护缓存、递归处理器等状态，复用同一实例才能命中缓存。每请求 json.New() 会丧失缓存收益并增加分配。

2. 共享实例安全，共享返回值容器要谨慎

Processor 可跨 goroutine 共享；但 Get 返回的 map/slice 若要在多 goroutine 间共享改写，调用方须自行加锁（或开启 CacheSharedResults 后视为只读）。

3. 用 Close 释放资源

长跑服务中显式 defer processor.Close() 与 defer iter.Close()，避免缓存 goroutine 与内存泄漏。SetGlobalProcessor 设置的实例改用 ShutdownGlobalProcessor。

4. CPU 密集型才值得并行

并行有调度与同步开销。小数组（< ParallelThreshold 默认 10）串行更快；JSONL 行数大、单行处理重时并行收益明显。

5. 并行模式注意行序

StreamJSONLParallel 不保证处理顺序。需要保序时按 lineNum 写入对位索引，处理完再顺序消费。

相关

• 性能优化 — 处理器复用、通用 Go 并发模式、基准测试
• 迭代器类型 — ParallelIterator 完整 API
• JSONL 处理 — 并行 JSONL API 详情
• 缓存与预解析 — 缓存机制与 PreParse 预解析
• 错误处理 — ErrConcurrencyLimit 等错误分类