Go语言中大型数据流的JSON渐进式编码实践

package main

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

func main() {
    type Data struct {
        Foo string
        Bar chan string // 这是一个数据流，不希望一次性加载
    }

    data := Data{
        Foo: "Hello World",
        Bar: make(chan string),
    }

    // 模拟一个长时间的数据流
    go func() {
        defer close(data.Bar)
        for _, x := range []string{"one", "two", "three", "four", "five"} {
            data.Bar <- x
            time.Sleep(100 * time.Millisecond) // 模拟数据生成延迟
        }
    }()

    // 尝试直接编码，这将导致错误
    if err := json.NewEncoder(os.Stdout).Encode(&data); err != nil {
        log.Println("Error encoding:", err) // 输出: Error encoding: json: unsupported type: chan string
    }
}

package main

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

// Data结构体包含一个固定字段和一个数据流字段
type Data struct {
    Foo string
    Bar chan string
}

// StreamEncodeJSON 函数用于流式编码Data结构体
func StreamEncodeJSON(w io.Writer, data Data) error {
    // 写入JSON对象的起始部分和Foo字段
    _, err := fmt.Fprintf(w, `{"Foo": %s, "Bar": [`, strconv.Quote(data.Foo))
    if err != nil {
        return fmt.Errorf("failed to write initial JSON: %w", err)
    }

    firstElement := true
    // 迭代通道，逐个编码Bar中的元素
    for item := range data.Bar {
        if !firstElement {
            // 如果不是第一个元素，先写入逗号分隔符
            _, err := w.Write([]byte(","))
            if err != nil {
                return fmt.Errorf("failed to write comma: %w", err)
            }
        }

        // 使用json.NewEncoder编码单个元素
        // 注意：json.Encoder.Encode 会在每个值后面添加换行符，
        // 但由于我们是手动拼接数组，需要确保输出是纯粹的JSON值。
        // 更稳妥的做法是使用 json.Marshal 再写入，或者确保Encoder不会写入换行符。
        // 对于基本类型如string，可以直接使用 strconv.Quote
        _, err = fmt.Fprintf(w, "%s", strconv.Quote(item))
        if err != nil {
            return fmt.Errorf("failed to encode item: %w", err)
        }
        firstElement = false
    }

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

func main() {
    data := Data{
        Foo: "Hello World",
        Bar: make(chan string),
    }

    // 模拟一个数据生成器
    go func() {
        defer close(data.Bar)
        for i := 0; i < 5; i++ {
            data.Bar <- fmt.Sprintf("element_%d", i+1)
            time.Sleep(100 * time.Millisecond) // 模拟数据生成延迟
        }
    }()

    fmt.Println("Starting stream encoding...")
    err := StreamEncodeJSON(os.Stdout, data)
    if err != nil {
        log.Fatalf("Stream encoding failed: %v", err)
    }
    fmt.Println("\nStream encoding finished.")
}

type Marshaler interface {
    MarshalJSON(io.Writer) error
}