当前位置：网站首页>Golang bufio Reader 源码详解

Golang bufio Reader 源码详解

2022-07-26 22:49:00 【itsgopher】

Golang — bufio Reader详解

bufio 是对 gosdk 种 io.Reader 和 io.writer 的二次打包，以 []byte 作为其 buffer，在 io.Reader 和 io.Writer 的基础上提供了更多方便有效的 io 读取和写入方法。

PS：由于代码过多，对源码函数流程的解释直接编写在源码注释上

io.Reader

io.Reader 和 io.writer 是 gosdk io 包中的非常重要的两个接口，它们内部各包含一个方法 Read 和 Write，传入参数 []byte ，返回读取/写入的字节数和错误。

type Reader interface {
    
	Read(p []byte) (n int, err error)
}

io.Reader核心作用为连续读取字节填充传入的 slice，并在读取停止时返回字节数和遇到的错误（无错误时返回nil）。

bufio Reader

bufio.Reader

// Reader implements buffering for an io.Reader object.
type Reader struct {
    
	buf          []byte
	rd           io.Reader // reader provided by the client
	r, w         int       // buf read and write positions
	err          error
	lastByte     int // last byte read for UnreadByte; -1 means invalid
	lastRuneSize int // size of last rune read for UnreadRune; -1 means invalid
}

该结构体是 bufio 种所有读取相关方法的 receiver，即所有读取操作都需要在创建一个此结构体类型的对象后才能使用。bufio.Reader 内部包含：

buf：读取的字节 buffer
rd：客户传入的 io.Reader 对象，即一切读取的字节都来自于该对象
r，w：读取和写入的标志位
err：当前发生的错误
lastByte：在 UnreadByte 方法中会使用到的上一个字节位置变量
lastRuneSize：在 UnreadRune 方法中会使用到的上一个 rune 位置变量

构造函数

构造函数包含两种，bufio.NewReader 和 bufio.NewReaderSize：

bufio.NewReaderSize：

func NewReaderSize(rd io.Reader, size int) *Reader {
      
	// 判断传入的rd是否已经是bufio.Reader类型的对象
	b, ok := rd.(*Reader)
	if ok && len(b.buf) >= size {
      
    // 如果是且其buf大于等于指定的size，直接返回
		return b
	}
  // 若size小于bufio中定义的minReadBufferSize，将其设为minReadBufferSize
  // minReadBufferSize = 16
	if size < minReadBufferSize {
      
		size = minReadBufferSize
	}
  // 利用new创建新的bufio.Reader对象
	r := new(Reader)
  // reset函数初始化bufio.Reader的所有域
  // 将buf设为大小为size的[]byte，将其中的io.Reader设为rd
	r.reset(make([]byte, size), rd)
	return r
}

bufio.NewReader：

func NewReader(rd io.Reader) *Reader {
      
  // 直接调用NewReaderSize，size设为默认的buffer size
  // defaultBufSize = 4096
	return NewReaderSize(rd, defaultBufSize)
}

bufio.Reader.fill

该方法是 bufio.Reader 非常核心的方法之一，它将暂存有效（还没有被读取）的 buffer 数据左滑倒初始位置，并多次尝试从 rd 中读取新的字节，只要读取到长度n大于0，错误等于nil的数据，将其append到buffer中。

// fill reads a new chunk into the buffer.
// 这里的 new chunk 的大小是不一定的
func (b *Reader) fill() {
    
  // 将当前有效的数据左滑到初始位置
	if b.r > 0 {
    
    // 左滑
		copy(b.buf, b.buf[b.r:b.w])
    // 改变读取和写入标志位
		b.w -= b.r
		b.r = 0
	}
	// 若当前写入标志位已达到buffer尾端（即buffer是满的状态），panic
	if b.w >= len(b.buf) {
    
		panic("bufio: tried to fill full buffer")
	}

  // 开始读取新的数据：尝试有限次，maxConsecutiveEmptyReads = 100，即尝试100次
	for i := maxConsecutiveEmptyReads; i > 0; i-- {
    
    // 从 rd 中读取
		n, err := b.rd.Read(b.buf[b.w:])
    // 读取数小于0是一个致命的错误，进行panic
		if n < 0 {
    
			panic(errNegativeRead)
		}
    // w 右滑n位
		b.w += n
    // 若出现了错误，将当前错误设为err并返回
		if err != nil {
    
			b.err = err
			return
		}
    // 若读取字节数大于0，结束循环返回
		if n > 0 {
    
			return
		}
    // 若读取字节数等于0，继续尝试
	}
  // 若尝试100次之后仍然没有新的字节，将当前错误设为ErrNoProgress，并返回
	b.err = io.ErrNoProgress
}

要注意的是，bufio 在遇到错误时不会摒弃当前读到的字节，会继续将其存下来，并更新当前的error。

bufio.Reader.ReadByte

该方法非常简单，从 buffer 中读取一个字节并返回即可。

// ReadByte reads and returns a single byte.
// If no byte is available, returns an error.
func (b *Reader) ReadByte() (byte, error) {
    
  // 因为当前操作为 ReadByte 就会使得 UnreadRune 无效，所以需要将 lastRuneSize 设为-1
	b.lastRuneSize = -1
  // b.r == b.w 代表当前没有可以读取的字节
  // 这里反复比较 b.r 和 b.w，保证在读取时会有可读取的字节
	for b.r == b.w {
    
    // 若具有错误，返回0和其错误
		if b.err != nil {
    
			return 0, b.readErr()
		}
    // 调用fill方法读取新的字节
		b.fill() // buffer is empty
	}
  // 读取新的字节
	c := b.buf[b.r]
  // b.r 右滑一位
	b.r++
  // 此时调用 UnreadByte 是有效的，将 lastByte 设为 int(c)
	b.lastByte = int(c)
	return c, nil
}

注意 b.readErr 操作会读取当前的err并返回，同时会把当前的err设为nil：

func (b *Reader) readErr() error {
    
	err := b.err
	b.err = nil
	return err
}

bufio.Reader.UnreadByte

UnreadByte 会对读取字节的操作进行一个撤销操作，源码中是这么解释的：

// UnreadByte unreads the last byte. Only the most recently read byte can be unread.
//
// UnreadByte returns an error if the most recent method called on the
// Reader was not a read operation. Notably, Peek, Discard, and WriteTo are not
// considered read operations.

即只有在最近的操作能被撤销的时候，才能正常执行，否则会返回错误。

func (b *Reader) UnreadByte() error {
    
  // 操作无效的两种情况：
  // 1. 上一次操作不是可撤销的操作利用 b.lastByte 去判断
  // 2. 读标志位在初始位置，而写标志位已右移
	if b.lastByte < 0 || b.r == 0 && b.w > 0 {
    
		return ErrInvalidUnreadByte
	}
	// b.r > 0 || b.w == 0
	if b.r > 0 {
    
    // 这种情况下，只需将读标志位撤回一位，将上一个字节填充即可
		b.r--
	} else {
    
		// b.r == 0 && b.w == 0
    // 这种情况，将写标志位右移一位，将上一个字节填充到0位置即可
		b.w = 1
	}
  // 填充字节
	b.buf[b.r] = byte(b.lastByte)
  // 不能进行任何其他撤销的动作
	b.lastByte = -1
	b.lastRuneSize = -1
	return nil
}

bufio.Reader.ReadRune & bufio.Reader.UnreadRune

该两种操作于 ReadByte 和 UnreadByte 基本类似，只是在读取的时候会有一些 UTF-8 编码到的问题，并且读取的不一定只是一个字节。

PS：关于 Rune 我以后还会专门写文档去介绍

bufio.Reader.Peek

Peek 顾名思义是一种 “瞄” 的操作，“只是去看看，不做任何变化”。Peek 会返回你指定个数的字节但是并不对 Reader 的内部状态做变化。其内部也有一些错误机制，在下面的源代码中进行解释：

func (b *Reader) Peek(n int) ([]byte, error) {
    
	// 字节数肯定不能小于0
  if n < 0 {
    
		return nil, ErrNegativeCount
	}
	// Peek 操作不能算是合法的读操作，所以不能进行撤销的操作
	b.lastByte = -1
	b.lastRuneSize = -1
	// 循环对 buffer 填充新的字节，直到以下条件中的任何一个满足：
  // 1. 读取了 n 个字节
  // 2. 字节数大于了 buffer size
  // 3. 发生读取错误
	for b.w-b.r < n && b.w-b.r < len(b.buf) && b.err == nil {
    
		b.fill() // b.w-b.r < len(b.buf) => buffer is not full
	}
	// 若 n 大于 buffer size：
  // 返回 r 到 w 的所有字节，和 buffer full 的错误
  // 即使不发生读取错误，也要解释为什么没有读到 n 个字节
	if n > len(b.buf) {
    
		return b.buf[b.r:b.w], ErrBufferFull
	}

	// 0 <= n <= len(b.buf)
	var err error
  // 若读取的字节数 avail 小于 n，将 n 设为 avail
  // 若没有发生读取错误，将 err 设为 ErrBufferFull
	if avail := b.w - b.r; avail < n {
    
		// not enough data in buffer
		n = avail
		err = b.readErr()
		if err == nil {
    
			err = ErrBufferFull
		}
	}
  // 返回 r 到 r+n 的所有字节和 err
	return b.buf[b.r : b.r+n], err
}

bufio.Reader.Discard

Discard 即摒弃，其可以让我们指定摒弃未来的多少个字节。首先我们来看看一个方法 Buffered：

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

明显能看出来，这个方法返回的是当前缓存的字节的数量即当前可以读取的字节的数量。

其次，我们再来看看 Discard 方法的实现：

func (b *Reader) Discard(n int) (discarded int, err error) {
    
   // 明显，要摒弃的字节数不能小于0
   if n < 0 {
    
      return 0, ErrNegativeCount
   }
   // 也不能等于0，只是等于0没有错误返回
   if n == 0 {
    
      return
   }
	 // Discard 跟 Peek 一样不是合法的读取操作，不能进行撤销
   b.lastByte = -1
   b.lastRuneSize = -1
	 // remain：当前剩余需要Discard的字节的数量
   remain := n
   // 开始循环
   for {
    
      // 获取当前可以读取的字节数为skip
      skip := b.Buffered()
      // 若 skip == 0，fill填充buffer
      if skip == 0 {
    
         b.fill()
         // 填充后再次获取当前可以读取的字节数更新skip
         skip = b.Buffered()
      }
     	// skip 应等于skip和remain中较小的一个
      if skip > remain {
    
         skip = remain
      }
     	// 读标志位向前移skip位
      b.r += skip
     	// remain减少skip
      remain -= skip
     	// remain==0 代表已摒弃n个字节，可以正常返回
      if remain == 0 {
    
         return n, nil
      }
     	// 如果发生错误
      // 返回摒弃的字节数，和错误
      if b.err != nil {
    
         return n - remain, b.readErr()
      }
   }
}

bufio.Reader.ReadSlice

ReadSlice 接受一个字节为参数，在buffer中寻找该字节，返回该字节之前的所有可读字节（以slice的形式）。源码中是这么解释 ReadSlice 的：

// ReadSlice reads until the first occurrence of delim in the input,
// returning a slice pointing at the bytes in the buffer.
// The bytes stop being valid at the next read.
// If ReadSlice encounters an error before finding a delimiter,
// it returns all the data in the buffer and the error itself (often io.EOF).
// ReadSlice fails with error ErrBufferFull if the buffer fills without a delim.
// Because the data returned from ReadSlice will be overwritten
// by the next I/O operation, most clients should use
// ReadBytes or ReadString instead.
// ReadSlice returns err != nil if and only if line does not end in delim.

ReadSlice 将参数 delim 作为边界读取字节，返回一个属于buffer的byte slice。返回值 slice 将在下一个读取操作后变为无效（因为其底部于buffer共享数组）。如果在遇到 delim 之前遇到错误，将返回所有数据和遇到的错误。如果在遇到 delim 之前 buffer 填满，将返回所有数据和 ErrBufferFull 错误。因为返回的数据有效期不长，所以不推荐用户使用该接口，推荐使用 ReadBytes 或 ReadString。

func (b *Reader) ReadSlice(delim byte) (line []byte, err error) {
    
	// 开始搜索的下标
  s := 0 // search start index
  // 开始循环
	for {
    
		// 利用 bytes.IndexByte 去在 buffer 中寻找 delim
    // 该方法在 delim 存在时返回 delim 下标，否则返回 -1
    // 如果返回值 i >= 0，代表当前的 buffer 中存在 delim，
    // 此时返回 buffer 中从 r 到 r+i+1 之间的字节即可
		if i := bytes.IndexByte(b.buf[b.r+s:b.w], delim); i >= 0 {
    
			i += s
			line = b.buf[b.r : b.r+i+1]
			b.r += i + 1
			break
		}
    
		// 若上述操作中 i == -1

		// 是否具有读取错误
		if b.err != nil {
    
      // 若是，返回所有数据和读取到的错误
			line = b.buf[b.r:b.w]
			b.r = b.w
			err = b.readErr()
			break
		}

		// buffer是否已填满
		if b.Buffered() >= len(b.buf) {
    
      // 若是，返回所有数据和ErrBufferFull
			b.r = b.w
			line = b.buf
			err = ErrBufferFull
			break
		}
		// s更新
		s = b.w - b.r // 不要去扫描已扫描过的区域
		// 填充buffer
		b.fill() // buffer is not full
	}

	// 因为该操作是一个合法的读操作
  // 在读取到字节之后，需要设置 lastByte 域
  // 使得可以进行UnreadByte操作
	if i := len(line) - 1; i >= 0 {
    
		b.lastByte = int(line[i])
		b.lastRuneSize = -1
	}

	return
}

bufio.Reader.collectFragments

此方法是一个在 ReadSlice 和其他 ReadBytes、ReadString等方法之间工作的中间方法。其循环调用 ReadSlice 方法直到遇到 delim 为止，并在过程中记录获取的所有 slice。该方法有四个返回值分别是：

fullBuffers：一个二维的字节slice，每次发生buffer full错误时，对buffer进行一次copy，并存到fullBuffers中。
finalFragment：在遇到 delim 之前读到的最后一串字节
totalLen：读取的总长度
err：发生的错误

func (b *Reader) collectFragments(delim byte) (fullBuffers [][]byte, finalFragment []byte, totalLen int, err error) {
    
	var frag []byte
  // 开始循环
	for {
    
		var e error
    // 调用 ReadSlice 方法
		frag, e = b.ReadSlice(delim)
    // 若返回没有错误，意味着读取到了 delim 字节，断开循环
    // 此时的 frag 就等于最终的 finalFragment
		if e == nil {
     // got final fragment
			break
		}
    // 若返回的错误不等于buffer已满，意味着发生了读取错误，停止循环
		if e != ErrBufferFull {
     // unexpected error
			err = e
			break
		}

		// 到这，就代表没有读取到 delim 并且 buffer 已填满
    // 此时，我们需要对 buffer 做一个复制，将其存到 fullbuffers 当中
		buf := make([]byte, len(frag))
		copy(buf, frag)
		fullBuffers = append(fullBuffers, buf)
		totalLen += len(buf)
	}
	// 在最终长度上添加 finalFragment 的长度
	totalLen += len(frag)
	return fullBuffers, frag, totalLen, err
}