本文介绍如何通过实现 io.reader 接口创建可组合、内存友好的自定义 reader，用于在不加载全部数据到内存的前提下，对大文件流（如 >10gb 的导入数据）实时过滤注释行、剔除非法字符等预处理操作。
本文介绍如何通过实现 io.reader 接口创建可组合、内存友好的自定义 reader，用于在不加载全部数据到内存的前提下，对大文件流（如 >10gb 的导入数据）实时过滤注释行、剔除非法字符等预处理操作。
在处理超大规模文本数据（例如 10GB+ 的批量导入文件）时，调用 ioutil.ReadAll 或 bytes.Buffer 全量读取会引发严重的内存压力甚至 OOM。Go 的 io.Reader 接口设计天然支持“流式处理”——我们无需修改原始数据源，而是构建一个中间过滤层 Reader，将原始文件 *os.File 封装后，按需提供已清洗的数据字节流。该 Reader 可无缝对接 pg.CopyFrom 等期望 io.Reader 输入的数据库接口。
核心思路是：实现 Read(p []byte) (n int, err error) 方法，在每次读取请求中完成逻辑过滤，并维护内部状态（如当前行缓冲、扫描位置）。下面是一个生产就绪的示例：
package main
import (
"bufio"
"io"
"strings"
)
// FilteredReader 是一个 io.Reader，用于跳过以 commentPrefix 开头的行，
// 并移除所有 badChar 字符，适用于 COPY FROM 流式导入场景。
type FilteredReader struct {
scanner *bufio.Scanner
comment string
badChar byte
buf []byte // 内部缓冲区，暂存已处理但未读出的字节
}
// NewFilteredReader 创建新的过滤 Reader
func NewFilteredReader(r io.Reader, comment, badChar string) *FilteredReader {
return &FilteredReader{
scanner: bufio.NewScanner(r),
comment: comment,
badChar: badChar[0], // 假设 badChar 为单字节（如 \x02）
buf: make([]byte, 0, 4096),
}
}
// Read 实现 io.Reader 接口：按需填充 p，返回实际写入字节数
func (fr *FilteredReader) Read(p []byte) (n int, err error) {
// 若内部缓冲有剩余数据，优先拷贝
if len(fr.buf) > 0 {
n = copy(p, fr.buf)
fr.buf = fr.buf[n:]
return n, nil
}
// 否则扫描下一行（跳过注释行）
var line []byte
for fr.scanner.Scan() {
line = fr.scanner.Bytes()
// 跳过以 comment 开头的整行（注意：需考虑空格/制表符前导）
if len(line) >= len(fr.comment) && strings.HasPrefix(string(line), fr.comment) {
continue
}
break
}
if err = fr.scanner.Err(); err != nil {
return 0, err
}
if len(line) == 0 { // 扫描结束
return 0, io.EOF
}
// 过滤 badChar 并追加换行符（确保行完整性，适配 COPY 协议）
cleaned := make([]byte, 0, len(line)+1)
for _, b := range line {
if b != fr.badChar {
cleaned = append(cleaned, b)
}
}
// 注意：原始文件可能含 \r\n 或 \n，此处保留原换行格式
// 如需统一为 \n，可在此处标准化：cleaned = bytes.ReplaceAll(cleaned, []byte("\r\n"), []byte("\n"))
cleaned = append(cleaned, '\n') // COPY 要求每行以换行结束
fr.buf = cleaned
return fr.Read(p) // 递归读取刚缓存的内容
}使用时，只需将原始文件封装进 FilteredReader，再传给 db.CopyFrom：
file, err := os.Open("/path/to/file")
if err != nil {
log.Fatal("open file:", err)
}
defer file.Close()
filtered := NewFilteredReader(file, "#", "\x02")
_, err = db.CopyFrom(filtered,
fmt.Sprintf("COPY %s FROM STDIN WITH DELIMITER E'%s'", dbName, "\x01"))
if err != nil {
log.Fatal("copy from:", err)
}⚠️ 关键注意事项：
- 换行符兼容性：bufio.Scanner 默认按 \n、\r\n、\r 切分，但 COPY 协议对行尾敏感。建议在 cleaned 后显式追加 '\n'，或根据源文件实际格式调整；
- 多字节 badChar：当前示例假设 badChar 为单字节（如 \x02）。若需处理 UTF-8 字符或多字节序列，应改用 strings.ReplaceAll + []byte 转换，但需权衡性能；
- 内存控制：FilteredReader.buf 容量按需增长，避免预分配过大；对超长行可添加长度限制（如 if len(line) > 1e6 { return 0, errors.New("line too long") }）；
- 错误传播：务必检查 scanner.Err() 并返回，否则 I/O 错误会被静默忽略；
- 组合性扩展：可嵌套多个 FilteredReader（如先去注释、再去控制字符、再转编码），体现 Go 接口组合哲学。
这种基于 io.Reader 的流式过滤模式，不仅内存高效、逻辑清晰，更与 Go 生态高度契合——它复用了标准库的抽象（如 bufio.Reader、gzip.Reader），使数据处理管道具备可插拔、可测试、可复用的工业级特性。
