Go 使用 json.Decoder 流式解析 NDJSON

功能说明

streamNDJSON 接收任意 io.Reader，用 json.Decoder 在循环中逐条解码 Event。每条记录校验 ID 为正、type 非空、amount_cents 非负，再交给回调聚合。示例用字符串模拟 NDJSON 输入，统计 credit 与 debit 数量以及净额；第二段输入缺少 type，展示错误发生前已处理的事件不会被丢弃。

关键代码

decoder.Decode 在流结束时返回 io.EOF，代码用 errors.Is 判断后正常返回。金额字段使用 int64 cents 而不是浮点数，避免货币累计误差。回调返回错误时立即停止，使业务校验可以中断整个流。json.Decoder 还支持连续拼接的 JSON 值，因此不强制每条记录后换行，但 NDJSON 约定更利于排查。解码器按需读取，内存占用与单条记录大小相关，而非文件总大小。

使用注意

该函数遇到第一条非法记录就返回，之前的聚合结果由调用方通过闭包保留；若需要跳过坏行，应改用 bufio.Scanner 按行切分并单独解码，同时设置最大行长。生产环境应从 http.Request.Body 读取并对总字节数设上限，防止超大记录。金额建议在数据库层用定点数或整数，展示时再格式化。不要用 float64 承载货币。

验证方式

运行 main，第一段输出 counts=map[credit:2 debit:1] net_cents=1825 err=；第二段输出 seen=1，并返回 event 5: type is required。可增加一条 amount_cents 为负的记录验证金额校验，或用 httptest 构造分块传输的响应，确认解码器在数据到达时逐条处理而不是等待 EOF。

代码示例

package main

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"strings"
)

type Event struct {
	ID          int               `json:"id"`
	Type        string            `json:"type"`
	AmountCents int64             `json:"amount_cents"`
	Meta        map[string]string `json:"meta,omitempty"`
}

func streamNDJSON(r io.Reader, handle func(Event) error) error {
	decoder := json.NewDecoder(r)
	for {
		var event Event
		if err := decoder.Decode(&event); err != nil {
			if errors.Is(err, io.EOF) {
				return nil
			}
			return fmt.Errorf("decode event: %w", err)
		}
		if event.ID <= 0 {
			return fmt.Errorf("event %d: id must be positive", event.ID)
		}
		if event.Type == "" {
			return fmt.Errorf("event %d: type is required", event.ID)
		}
		if event.AmountCents < 0 {
			return fmt.Errorf("event %d: amount_cents must be >= 0", event.ID)
		}
		if err := handle(event); err != nil {
			return err
		}
	}
}

func main() {
	payload := strings.Join([]string{
		`{"id":1,"type":"credit","amount_cents":1250}`,
		`{"id":2,"type":"debit","amount_cents":325,"meta":{"channel":"api"}}`,
		`{"id":3,"type":"credit","amount_cents":900}`,
	}, "\n")

	var netCents int64
	counts := map[string]int{}
	err := streamNDJSON(strings.NewReader(payload), func(event Event) error {
		counts[event.Type]++
		if event.Type == "credit" {
			netCents += event.AmountCents
		} else {
			netCents -= event.AmountCents
		}
		return nil
	})
	fmt.Printf("counts=%v net_cents=%d err=%v\n", counts, netCents, err)

	bad := bytes.NewBufferString("{\"id\":4,\"type\":\"credit\",\"amount_cents\":500}\n{\"id\":5,\"amount_cents\":100}\n")
	seen := 0
	err = streamNDJSON(bad, func(Event) error {
		seen++
		return nil
	})
	fmt.Printf("seen=%d err=%v\n", seen, err)
}