目录
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.Readerr := 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 []byterd io.Reader // reader provided by the clientr, w int // buf read and write positionserr errorlastByte int // last byte read for UnreadByte; -1 means invalidlastRuneSize 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 = 0b.w = 0n, 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 = 8b.r += nb.lastByte = int(b.buf[b.r-1])b.lastRuneSize = -1return 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 = -1b.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 errorif avail := b.w - b.r; avail < n {// not enough data in buffern = availerr = 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.rb.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 += nif err != nil {b.err = errreturn}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, 40fmt.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:4056the 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 errorbuf []byten intwr 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 -= nb.err = errreturn err}b.n = 0return 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) {
//先判断是否在EOFif 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 = tokenif 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使用了缓存的策略。)
缓冲区什么时候发送由内核决定,缓冲区也不能无线大,因为用户缓冲区发送到内核缓冲区再发送到网卡,网卡一次发出去的数据有最大长度,不管累计多少最后还是分片发送,这样缓冲区太大没有意义,数据传输也是有延时要求的,不能总在缓冲区呆着,太大也浪费资源