Question

平台无关的缓冲 I/O，提升读写效率。

package main

import (
	// "bufio"
	"io"
	"os"
)

func main() {
	f, _ := os.Open("./tmp.dat")
	defer f.Close()

	var r io.Reader = f
	// r = bufio.NewReaderSize(r, 8192)

	for {
		buf := make([]byte, 512)
		 _, err := r.Read(buf)
		if err == io.EOF { break }
	}
}

/*

$ dd if=/dev/random of=tmp.dat bs=1M count=100
$ strace ./test 2>&1 | grep "read" | wc -l

*/

源码剖析

默认缓冲区 4 KB，可用 NewReaderSize 指定。

// bufio/bufio.go

type Reader struct {
	buf          []byte
	rd           io.Reader // reader provided by the client
	r, w         int       // buf read and write positions
	err          error
}

const defaultBufSize = 4096
const minReadBufferSize = 16

func NewReader(rd io.Reader) *Reader {
	return NewReaderSize(rd, defaultBufSize)
}

func NewReaderSize(rd io.Reader, size int) *Reader {
    
	// Is it already a Reader?
	b, ok := rd.(*Reader)
	if ok && len(b.buf) >= size {
		return b
	}
    
	if size < minReadBufferSize {
		size = minReadBufferSize
	}
    
	r := new(Reader)
	r.reset(make([]byte, size), rd)
	return r
}

读操作，需看有没有必要用缓冲区。
即便从缓冲区取数据，也是有多少读多少。

func (b *Reader) Buffered() int { return b.w - b.r }


func (b *Reader) Read(p []byte) (n int, err error) {
    
    // 缓冲区为空。
	if b.r == b.w {
		if b.err != nil {
			return 0, b.readErr()
		}
        
		if len(p) >= len(b.buf) {
            
            // Large read, empty buffer.
            // Read directly into p to avoid copy.
            
			n, b.err = b.rd.Read(p)            
			return n, b.readErr()
		}
        
        // 从底层 Reader 读取数据，填充缓冲区。
        // 缓冲区 4KB(?)，而用户 Read 量可能少得多，
        // 如此，可避免多次底层读操作。
		b.r = 0
		b.w = 0
		n, b.err = b.rd.Read(b.buf)
		if n == 0 {
			return 0, b.readErr()
		}
		b.w += n
	}

    // 自缓冲区拷贝数据，有多少算多少。
	n = copy(p, b.buf[b.r:b.w])
	b.r += n
    
	return n, nil
}

填充操作。先将缓冲区遗留数据前移，再尝试（次数限制）填满后续空间。

func (b *Reader) fill() {
    
	// 将缓冲区数据前移。
	if b.r > 0 {
		copy(b.buf, b.buf[b.r:b.w])
		b.w -= b.r
		b.r = 0
	}

	// Read new data: try a limited number of times.
	for i := maxConsecutiveEmptyReads; i > 0; i-- {
        
        // 自底层 Reader 读取数据填满缓冲区。
		n, err := b.rd.Read(b.buf[b.w:])
		b.w += n
        
		if err != nil {
			b.err = err
			return
		}
        
        // 成功读取，不再尝试。
		if n > 0 {
			return
		}
	}
    
	b.err = io.ErrNoProgress
}

重置操作，简单恢复计数状态。

func (b *Reader) Reset(r io.Reader) {
	if b.buf == nil {
		b.buf = make([]byte, defaultBufSize)
	}
	b.reset(b.buf, r)
}

func (b *Reader) reset(buf []byte, r io.Reader) {
	*b = Reader{
		buf:          buf,
		rd:           r,
		lastByte:     -1,
		lastRuneSize: -1,
	}
}

写操作过程，大体类似。

type Writer struct {
	err error
	buf []byte
	n   int
	wr  io.Writer
}

func NewWriterSize(w io.Writer, size int) *Writer {
    
	// Is it already a Writer?
	b, ok := w.(*Writer)
	if ok && len(b.buf) >= size {
		return b
	}
    
	if size <= 0 {
		size = defaultBufSize
	}
    
	return &Writer{
		buf: make([]byte, size),
		wr:  w,
	}
}

func NewWriter(w io.Writer) *Writer {
	return NewWriterSize(w, defaultBufSize)
}

func (b *Writer) Available() int { return len(b.buf) - b.n }


func (b *Writer) Write(p []byte) (nn int, err error) {
    
    // 待写入数据量大于缓冲区剩余空间，多次完成。
	for len(p) > b.Available() && b.err == nil {
		var n int
        
		if b.Buffered() == 0 {
            
            // Large write, empty buffer.
            // Write directly from p to avoid copy.
            
			n, b.err = b.wr.Write(p)
		} else {
            
            // 填满剩余缓冲区空间。
			n = copy(b.buf[b.n:], p)
			b.n += n
            
            // 将缓冲区数据 Flush 给底层 Writer。
            // 清空缓冲区。
			b.Flush()
		}
		nn += n
		p = p[n:]
	}

    // 复制数据到缓冲区。
    // 1. 上面循环从未执行。
    // 2. 最后一次循环，剩余量不足以继续。
	n := copy(b.buf[b.n:], p)
	b.n += n
	nn += n
    
	return nn, nil
}

func (b *Writer) Flush() error {
	n, err := b.wr.Write(b.buf[0:b.n])
    
	if err != nil {
        // 部分写入。剩余数据前移。
		if n > 0 && n < b.n {
			copy(b.buf[0:b.n-n], b.buf[n:b.n])
		}
        
		b.n -= n
		b.err = err
		return err
	}
    
    // 全部写入。
	b.n = 0
	return nil
}