【Go】-bufio库解读

目录

Reader和Writer接口

bufio.Reader/Writer

小结

其他函数-Peek、fill

Reader小结

Writer

Scanner结构体

缓冲区对于网络数据读写的重要性

Reader和Writer接口

        在net/http包生成的Conn 接口的实例中有两个方法叫做Read和Write接口

type Conn interface {
    Read(b []byte)(n int,err error)
    Write(b []byte)(n int,err error)
    Close() error
    ...
}

        该接口实际上实现了io.Reader和io.Write的方法

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

type Writer interface {
    Write(p []byte)(n int, err error)
}

        在Unix中，一切皆文件，Golang 把这个思想贯彻到更远，因为本质上我们对文件的抽象就是一个可读可写的一个对象，也就是实现了io.Writer 和 io.Reader 的对象我们都可以称为文件。
其中我们需要注意Reader接口的几个细节：

• Reader.Read函数最多读取len(p)字节的数据保存到P中，若n<len(p)则立即返回
• 在输入流结束时，方法会返回n>0字节，但是error可能是EOF也可能是nil，在这种情况下再次调用read方法肯定会返回(0,EOF)
• 调用read方法，当n>0要优先处理读入数据再处理err---即使我们在读取的时候遇到错误，但是也应该也处理已经读到的数据，因为这些已经读到的数据是正确的，如果不进行处理丢失的话，读到的数据就不完整了。

bufio.Reader/Writer

        那么我们对Socket读写的行为就可以巨象成对Conn的Read和Write。

        demo.go

func main() {
    str := strings.Repeat("123", 20)
    reader := strings.NewReader(str)
    fmt.Printf("the size of strings reader is %d\n", reader.Size())
    fmt.Println("create a new reader")
    var buffer *bufio.Reader = bufio.NewReader(reader)
    fmt.Printf("the default size of buffered reader is %d\n", buffer.Size())
    fmt.Printf("The number of unread bytes in the buffer: %d\n", buffer.Buffered())
    buf1 := make([]byte, 8)
    for {
        n, err := buffer.Read(buf1)
        fmt.Printf("read the number of Context is %d\n", n)
        fmt.Print("the context is")
        fmt.Println(string(buf1[:n]))
        if err == io.EOF {
            break
        }
        if err != nil {
            log.Fatal(err)
        }
        fmt.Printf("The number of unread bytes in the buffer: %d\n", buffer.Buffered())
    }
    n, err := buffer.Read(buf1)
    fmt.Println("try again!")
    fmt.Println(n, err)
}

        在该例子中
        1.创建了一个长度为60字节的字符串（因为数字对应的ASCII码长度为1字节），利用该字符串创建了 bufio.Reader为 buffer（default size为4096字节，而且初始化时当前缓冲区可读取的字节数为0）

        2.创建了一个长度为8的切片用于循环读取
        3.直到读取到EOF为止退出循环
        4.当尝试再次读取时失败，数据流被设计成单向（是不是意味着跟读取文件一样，内部有一个读指针表示已读且不重置？）

        我们将断点打在bufio.NewReader尝试通过源码解释这现象

dlv debug demo.go
b bufio.NewReader

        newReader实际上是对NewReaderSize的包装，默认缓冲区大小为defaultBufSize，defaultBufSize = 4096

                                                ​​​​​​​        ​​​​​​​        

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就是*bufio.Reader
        r := new(Reader)
        r.reset(make([]byte, size), rd)
        return r
}

        如果我们创建io.Reader时的字符串比4096还要长，那么我们直接返回那个io.Reader。当然缓冲区也有最小长度要求minReadBufferSize = 16。之后的new关键词我们也知道，底层调用mallocgc分配内存，reset函数是重置自身所有字段

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

        可以看到buf缓冲区实际上就是一个size长度的切片，我们回头再看下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
}

• buf 知道了是缓冲区切片
• rd 是我们在调用NewReader时传入的io.Reader，也就是说他底层保存着原始数据的拷贝，是一个底层读取器，那是不是意味着缓冲区要从这里面读取数据？同时也解释了为啥要返回一个指针，因为拷贝传值占用太大了
• r和w 是代表缓冲区进行下一次读/写的开始，是已读/已写指针，看到这个我们就明白了如何被设计成单向数据流
• lastByte/lastRuneSize 用于记录缓冲区最后一个被读的字节和码点，读回退时用到

        我们为了了解buf和rd的关系，继续给Read打断点

r
b bufio.Reader.Read

Read函数

func (b *Reader) Read(p []byte) (n int, err error) {
//首先确定被写入切片长度
        n = len(p)
        if n == 0 {
                if b.Buffered() > 0 {
                        return 0, nil
                }
                return 0, b.readErr()
        }
//一开始肯定是0进入该分支
        if b.r == b.w {
//EOF时也会触发这个
                if b.err != nil {
                        return 0, b.readErr()
                }
//当被写入切片长度远大于缓冲区
//直接从原始io.Reader整个读入
                if len(p) >= len(b.buf) {
                        // Large read, empty buffer.
                        // Read directly into p to avoid copy.
                        n, b.err = b.rd.Read(p)
                        if n < 0 {
                                panic(errNegativeRead)
                        }
                        if n > 0 {
                                b.lastByte = int(p[n-1])
                                b.lastRuneSize = -1
                        }
                        return n, b.readErr()
                }
                // One read.
                // Do not use b.fill, which will loop.
//r=w其中一种情况就是第一次读取
//从原始io.Reader中即rd读取buf
//之后写指针往前推移
                b.r = 0
                b.w = 0
                n, b.err = b.rd.Read(b.buf)
                if n < 0 {
                        panic(errNegativeRead)
                }
                if n == 0 {
                        return 0, b.readErr()
                }
//此时b.w = 60是io.Reader的数据最长长度
//也就是说下一次进入这里是读完的时候
                b.w += n
        }
// n 为传入的切片长度
//拷贝进切片后读指针往前推移
        n = copy(p, b.buf[b.r:b.w])
//b.r = 8
        b.r += n
        b.lastByte = int(b.buf[b.r-1])
        b.lastRuneSize = -1
        return n, nil

        buf确实是从rd中读取数据，利用w和r确保已读计数之前的字节都被读取且不会再次被读，因此在切片上（demo中指buf1）是安全的，我们之前的buffer.Buffered的读取缓冲区实际上是w-r的差值，即buf中数据的长度减去已读的，刚开始两者都是0，所以一开始为0

        ​​​​​​​        

小结

其他函数-Peek、fill

        有个奇怪的地方就是Peek方法用于查看未读取数据的n个字节，但并不会改变bufio.Reader的状态

func (b *Reader) Peek(n int) ([]byte, error) {
        if n < 0 {
                return nil, ErrNegativeCount
        }
//设置为-1表示回退操作失败，即不能执行回退
        b.lastByte = -1
        b.lastRuneSize = -1
//未读取数据的长度小于n 且缓冲区未满
//执行fill填充缓冲区
        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 大于缓冲区长度就直接返回全部及错误提示
        if n > len(b.buf) {
                return b.buf[b.r:b.w], ErrBufferFull
        }

        // 0 <= n <= len(b.buf)
//有效数据长度小于n，表示fill没填满缓冲区，
//此时最多返回所有有效数据并产生错误提示
        var err error
        if avail := b.w - b.r; avail < n {
                // not enough data in buffer
                n = avail
                err = b.readErr()
                if err == nil {
                        err = ErrBufferFull
                }
        }
        return b.buf[b.r : b.r+n], err
}

        这个函数有一个问题就是return的时候直接返回切片，那意味着调用者可以直接修改缓冲区的值！造成数据泄露风险，当下次读取时r和w改变，当前数据位置也改变，慎用该函数
同理我们阅读文档发现ReadLine和ReadSlice有同样的返回类型

ReadLine() (line []byte,isPrefix bool,err error)
ReadSlice() (line []byte,err error)

        两者都会返回缓存切片，都存在内存泄漏问题

        fill函数

func (b *Reader) fill() {
        // Slide existing data to beginning.
//b.r>0表示已经读了一部分想要从0开始读就需要重置
//这里直接修改了缓冲区数据，进行数据平移
        if b.r > 0 {
                copy(b.buf, b.buf[b.r:b.w])
                b.w -= b.r
                b.r = 0
        }

        if b.w >= len(b.buf) {
                panic("bufio: tried to fill full buffer")
        }

        // Read new data: try a limited number of times.
// maxConsecutiveEmptyReads =100
//
        for i := maxConsecutiveEmptyReads; i > 0; i-- {
//从rd读取w位置数据到缓冲区
                n, err := b.rd.Read(b.buf[b.w:])
                if n < 0 {
                        panic(errNegativeRead)
                }
                b.w += n
                if err != nil {
                        b.err = err
                        return
                }
                if n > 0 {
                        return
                }
        }
        b.err = io.ErrNoProgress
}

        fill先查已读计数，大于0时有两种情况：

• 1.当有效数据长度大于无效数据长度（r已读的数据），copy可以直接覆盖
• 2.当有效数据长度小于无效数据长度，即不能完全用copy覆盖怎么办？无所谓反正是根据[b.r:b.w]来读取，后面舍弃就行，这就体现了r和w设计的妙处

        fill方法只要在开始时发现其所属值的已读计数大于0，就会对缓冲区进行一次压缩。之后，如果缓冲区中还有可写的位置，那么该方法就会对其进行填充。
        在填充缓冲区的时候，fill方法会试图从底层读取器那里，读取足够多的字节，并尽量把从已写计数w代表的索引位置到缓冲区末尾之间的空间都填满。

Reader小结

        Reader内部的缓冲区buf实际上是一个切片默认大小为4KB，它介于底层读取器rd和调用方之间，Reader渎职一般是先从底层读取器rd中读一部分数据（缓冲区足够大就全放入）放入缓冲区，再从buf中读取，并且从安全性方面考虑，Peek、ReadSlice和ReadLine方法都会造成内存泄漏问题，调用方可以直接修改缓冲区，这十分危险

Writer

func main() {
    str := strings.Repeat("hi", 100)
    basicWriter := &strings.Builder{}
    fmt.Printf("New a buffered writer writer size \n")
    writer1 := bufio.NewWriter(basicWriter)
    fmt.Println()
    fmt.Printf("the number of buffered bytes %d\n", writer1.Buffered())
    fmt.Printf("the number of unused bytes in the buffer:%d\n", writer1.Available())

    begin, end := 0, 40
    fmt.Printf("Write %d byte into the writer\n", end-begin)
    writer1.Write(([]byte(str))[begin:end])
    fmt.Printf("the number of buffered bytes %d\n", writer1.Buffered())
    fmt.Printf("the number of unused bytes in the buffer:%d\n", writer1.Available())
    fmt.Println()
    writer1.Flush()
    fmt.Printf("the number of buffered bytes %d\n", writer1.Buffered())
    fmt.Printf("the number of unused bytes in the buffer:%d\n", writer1.Available())

}

New a buffered writer writer size 

the number of buffered bytes 0
the number of unused bytes in the buffer:4096
Write 40 byte into the writer
the number of buffered bytes 40
the number of unused bytes in the buffer:4056

the number of buffered bytes 0
the number of unused bytes in the buffer:4096

        NewWriter和NewReader一样，默认给一个4k缓冲区

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,
        }
}

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

        Writer结构体字段没那么多：

• err存储报错信息
• buf 缓冲区
• n 已写指针，对缓冲区进行下一次写入的开始索引
• wr 底层写入器

函数方法的思想和Reader类似，这里不再赘述，需要注意的是Flush方法将缓冲区buf推入wr

func (b *Writer) Flush() error {
        if b.err != nil {
                return b.err
        }
        if b.n == 0 {
                return nil
        }
        n, err := b.wr.Write(b.buf[0:b.n])
        if n < b.n && err == nil {
                err = io.ErrShortWrite
        }
        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
}

        实际上是通过n进行的覆盖，这个方法在Write中也出现过                        
 

        为的是给后续新数据腾出空间，如果Write方法发现需要写入的字节太多，同时缓冲区已空，那么它就会跨过缓冲区，并直接把这些数据写到底层写入器中。
        总之，在通常情况下，只要缓冲区中的可写空间无法容纳需要写入的新数据，Flush方法就一定会被调用。不过，在你把所有的数据都写入Writer值之后，再调用一下它的Flush方法，显然是最稳妥的。

Scanner结构体

        从上文的分析中我们了解到读取一行ReadLine底层实际是通过ReadSlice('\n')实现，存在内存泄漏的问题。因此对于简单的读取一行，Reader中没有让人特别满意的方法，于是1.1增加了Scanner类型
简单demo

func main() {
    scanner := bufio.NewScanner(os.Stdin)
    for scanner.Scan() {
        fmt.Println(scanner.Text())
    }
    if err := scanner.Err(); err != nil {
        fmt.Fprintln(os.Stderr, "reading standard input:", err)
    }
}

        Scanner结构体(scan.go)

type Scanner struct {
        r            io.Reader // The reader provided by the client.
        split        SplitFunc // The function to split the tokens.
        maxTokenSize int       // Maximum size of a token; modified by tests.
        token        []byte    // Last token returned by split.
        buf          []byte    // Buffer used as argument to split.
        start        int       // First non-processed byte in buf.
        end          int       // End of data in buf.
        err          error     // Sticky error.
        empties      int       // Count of successive empty tokens.
        scanCalled   bool      // Scan has been called; buffer is in use.
        done         bool      // Scan has finished.
}

• r和buf是底层读取和缓冲,start和end大概率猜到是已读已写指针
• split、maxTokenSize、token需要往下看一下

split类型签名如下

type SplitFunc func(data []byte, atEOF bool) (advance int, token []byte, err error)

其中ScanBytes/Line/Runes/Words是其的具体实现

        默认使用ScanLines，且初始化时不创建缓冲区，maxTokenSize大小默认为64KB
实际上SplitFunc定义了对输入进行分词的slit签名，data是未处理数据，atEOF表示是否还有数据(EOF才算结尾)，advance表示读取字节数，token表示下一个结果数据

何为token?
        有数据 "studygolang\tpolaris\tgolangchina"，通过"\t"进行分词，那么会得到三个token，它们的内容分别是：studygolang、polaris 和 golangchina。而 SplitFunc 的功能是：进行分词，并返回未处理的数据中第一个 token。对于这个数据，就是返回 studygolang。

ScanLine源码

func ScanLines(data []byte, atEOF bool) (advance int, token []byte, err error) {
//先判断是否在EOF
        if atEOF && len(data) == 0 {
                return 0, nil, nil
        }
//判断有没有换行标志‘\n’，没有indexByte返回-1
//drop默认去掉结尾的'\r'
        if i := bytes.IndexByte(data, '\n'); i >= 0 {
                // We have a full newline-terminated line.
                return i + 1, dropCR(data[0:i]), nil
        }
        // If we're at EOF, we have a final, non-terminated line. Return it.
        if atEOF {
                return len(data), dropCR(data), nil
        }
        // Request more data.
        return 0, nil, nil
}

func dropCR(data []byte) []byte {
        if len(data) > 0 && data[len(data)-1] == '\r' {
                return data[0 : len(data)-1]
        }
        return data
}

        split会根据传入的函数类型截取数据流（Line就是截取直到'\n'，Byte就是截取一个，‘Words’截取直到'\t\n\v\f\r'，因为这几个都是空格unicode.IsSpace(),Rune截取码点），因此splite是一个分词策略，token就是这个分词策略所分出的词
我们可以使用Scanner.Split方法来更换分词策略

func (s *Scanner) Split(split SplitFunc) {
        if s.scanCalled {
                panic("Split called after Scan")
        }
        s.split = split
}

Scanner.Scan方法内部会循环直到获得一个分词
        伪代码：

func Sca(){
  for {
    advance, token, err := s.split(s.buf[s.start:s.end], s.err != nil)
    s.token = token
    if token!=nil {return true}
  }
}

        内部调用split方法存储结果在tokn中，当token=nil会移动已读已写指针Scanner.Text

func (s *Scanner) Text() string {
        return string(s.token)
}

        因为每执行一次Scan，token会被覆盖一次，因此需要使用for循环读取（注意buf缓冲区在Scan调用split的时有使用到）

缓冲区对于网络数据读写的重要性

        缓冲区对于接收方的意义在于减少程序压力，不至于被一次性的大量数据给压垮。对于发送方的意义就是节省带宽，一次性发送更多数据，也减少了用户态到内核态的切换（内核协议栈不确定用户一次要发多少数据，如果用户来一次就发一次，如果数据多还好说，如果少了，那网络I/O很频繁，而真正发送出去的数据也不多，所以为了减少网络I/O使用了缓存的策略。）
缓冲区什么时候发送由内核决定，缓冲区也不能无线大，因为用户缓冲区发送到内核缓冲区再发送到网卡，网卡一次发出去的数据有最大长度，不管累计多少最后还是分片发送，这样缓冲区太大没有意义，数据传输也是有延时要求的，不能总在缓冲区呆着，太大也浪费资源