Go语言流式JSON编码：处理大型数据集与Channel的实践策略

package main

import (
    "encoding/json"
    "log"
    "os"
)

type MyData struct {
    Foo string
    Bar chan string // 这是一个数据流，不应一次性加载
}

func main() {
    dataChan := make(chan string)
    t := MyData{
        Foo: "Hello World",
        Bar: dataChan,
    }

    go func() {
        defer close(dataChan)
        for _, x := range []string{"one", "two", "three"} {
            dataChan <- x
        }
    }()

    // 尝试直接编码将失败
    if err := json.NewEncoder(os.Stdout).Encode(&t); err != nil {
        log.Printf("错误示例: %v", err) // 输出: json: unsupported type: chan string
    }
}

package main

import (
    "encoding/json"
    "fmt"
    "io"
    "log"
    "os"
    "sync"
)

// MyStreamData 结构体，Bar字段是一个接收通道
type MyStreamData struct {
    Foo string
    Bar <-chan string // 使用接收通道，明确其流式特性
}

// StreamMarshalJSON 方法实现了流式JSON编码逻辑
func (s *MyStreamData) StreamMarshalJSON(w io.Writer) error {
    // 1. 写入JSON对象的起始部分和固定字段 (Foo)
    // 使用fmt.Fprintf确保字符串格式化和写入
    if _, err := fmt.Fprintf(w, `{ "Foo": "%s", "Bar": [`, s.Foo); err != nil {
        return fmt.Errorf("写入Foo字段失败: %w", err)
    }

    // 2. 迭代Channel，逐个编码流式数据
    firstItem := true
    for item := range s.Bar {
        // 在除第一个元素外的所有元素前添加逗号
        if !firstItem {
            if _, err := w.Write([]byte(",")); err != nil {
                return fmt.Errorf("写入逗号失败: %w", err)
            }
        }

        // 将单个元素编码为JSON字节，然后写入Writer
        // 注意：这里使用json.Marshal而非json.NewEncoder(w).Encode(item)
        // 是因为json.NewEncoder(w).Encode(item)会在每个元素后添加换行符，
        // 这会破坏JSON数组的格式。json.Marshal返回纯粹的JSON字节。
        itemBytes, err := json.Marshal(item)
        if err != nil {
            return fmt.Errorf("编码流式元素失败: %w", err)
        }
        if _, err := w.Write(itemBytes); err != nil {
            return fmt.Errorf("写入流式元素失败: %w", err)
        }
        firstItem = false
    }

    // 3. 写入JSON数组和对象的结束部分
    if _, err := w.Write([]byte("]}")); err != nil {
        return fmt.Errorf("写入JSON结束符失败: %w", err)
    }
    return nil
}

func main() {
    dataChan := make(chan string)
    streamData := MyStreamData{
        Foo: "Hello World",
        Bar: dataChan,
    }

    var wg sync.WaitGroup
    wg.Add(1)
    go func() {
        defer wg.Done()
        defer close(dataChan) // 确保Channel在所有数据发送后关闭
        items := []string{"one", "two", "three", "four", "five", "six", "seven", "eight", "nine", "ten"}
        for _, x := range items {
            dataChan <- x
            // 模拟数据生成延迟，观察流式输出效果
            // time.Sleep(50 * time.Millisecond)
        }
    }()

    log.Println("开始流式JSON编码...")
    if err := streamData.StreamMarshalJSON(os.Stdout); err != nil {
        log.Fatalf("流式JSON编码失败: %v", err)
    }
    fmt.Println() // 在输出末尾添加一个换行符，使终端显示更整洁
    log.Println("流式JSON编码完成。")

    wg.Wait() // 等待数据生成goroutine完成
}

// 假设这是 encoding/json 内部的某个处理函数片段
// 注意：这只是一个概念性示例，无法直接编译或在外部使用。
func (e *encoder) reflectValue(v reflect.Value) {
    switch v.Kind() {
    // ... 其他类型处理 ...

    case reflect.Array: // 数组的处理方式
        e.WriteByte('[')
        n := v.Len()
        for i := 0; i < n; i++ {
            if i > 0 {
                e.WriteByte(',')
            }
            e.reflectValue(v.Index(i))
        }
        e.WriteByte(']')

    case reflect.Chan: // 假设新增对Channel的处理
        e.WriteByte('[')
        i := 0
        for {
            // 尝试从Channel接收数据
            x, ok := v.Recv() // v.Recv() 是 reflect.Value 的方法，用于从Channel接收
            if !ok {
                break // Channel已关闭且无更多数据
            }
            if i > 0 {
                e.WriteByte(',')
            }
            // 递归调用自身编码接收到的元素
            e.reflectValue(x)
            i++
        }
        e.WriteByte(']')

    // ... 其他类型处理 ...
    }
}