结构
LimitedReader
定义
限制从Reader中读取的字节数。
type LimitedReader struct {
R Reader // underlying reader
N int64 // max bytes remaining
}
方法
//Limited
func (l *LimitedReader) Read(p []byte) (n int, err error) {
//已经读完了返回EOF
if l.N <= 0 {
return 0, EOF
}
//限制读取的长度
if int64(len(p)) > l.N {
p = p[0:l.N]
}
//直接调用Reader的Read
n, err = l.R.Read(p)
//更新N，保证一个LimitReader限制字节数
l.N -= int64(n)
return
}
SectionReader
定义
实现了对底层满足ReadAt接口的输入流某个片段的Read、ReadAt、Seek方法.
type SectionReader struct {
r ReaderAt
base int64 //读取的起始偏移
off int64 //读取时的指针
limit int64 //最终位置
}
方法
//读
func (s *SectionReader) Read(p []byte) (n int, err error) {
//超过limit返回EOF
if s.off >= s.limit {
return 0, EOF
}
//计算最大可以读取的长度，进而修改slice
if max := s.limit - s.off; int64(len(p)) > max {
p = p[0:max]
}
//从off开始读取len(p)个字节
n, err = s.r.ReadAt(p, s.off)
s.off += int64(n)
return
}
//跳转后的偏移能大于limit吗？？
//Seeker
//返回跳转后与起始位置的偏移
func (s *SectionReader) Seek(offset int64, whence int) (int64, error) {
switch whence {
default:
return 0, errWhence
case SeekStart: //SeekStart = 0 // 相对起始位置跳转
offset += s.base
case SeekCurrent: //SeekCurrent = 1 // 相对当前位置跳转
offset += s.off
case SeekEnd: //SeekEnd = 2 // 相对最后位置跳转
offset += s.limit
}
//跳转后偏移必须大于起始位置
if offset < s.base {
return 0, errOffset
}
s.off = offset
return offset - s.base, nil
}
//off为相对base偏移量
func (s *SectionReader) ReadAt(p []byte, off int64) (n int, err error) {
//不能大于Size()
if off < 0 || off >= s.limit-s.base {
return 0, EOF
}
//设置ReadAt的初始位置,该off不能影响SectionReader.off
off += s.base
//计算可以读取的长度
if max := s.limit - off; int64(len(p)) > max {
p = p[0:max]
n, err = s.r.ReadAt(p, off)
//此时len(p)>max,因此需要设置文件结束的err
if err == nil {
err = EOF
}
return n, err
}
//可以直接读取
return s.r.ReadAt(p, off)
}
func (s *SectionReader) Size() int64 { return s.limit - s.base }
teeReader
定义
私有结构，需要通过TeeReader函数创建一个teeReader，使读取r的数据之前均写入到w中
type teeReader struct {
r Reader
w Writer
}
方法
func (t *teeReader) Read(p []byte) (n int, err error) {
//从r中读数据
n, err = t.r.Read(p)
if n > 0 {
//想w写入数据
if n, err := t.w.Write(p[:n]); err != nil {
return n, err
}
}
return
}
multiReader
定义
通过MultiReader创建，从多个Reader中连续读取
type multiReader struct {
readers []Reader
}
方法
//实现Reader接口
//从readers中按顺序读取数据
func (mr *multiReader) Read(p []byte) (n int, err error) {
for len(mr.readers) > 0 {
// Optimization to flatten nested multiReaders (Issue 13558).
//嵌套multiReader
if len(mr.readers) == 1 {
if r, ok := mr.readers[0].(*multiReader); ok {
mr.readers = r.readers
continue
}
}
n, err = mr.readers[0].Read(p)
//判断当前Reader是否读完
if err == EOF {
//释放对已读完的Reader的引用
mr.readers[0] = eofReader{} // permit earlier GC
//剔除第一个元素
mr.readers = mr.readers[1:]
}
//如果readers还有则不能返回EOF
if n > 0 || err != EOF {
if err == EOF && len(mr.readers) > 0 {
// Don't return EOF yet. More readers remain.
err = nil
}
return
}
}
return 0, EOF
}
multiWriter
定义
通过MultiWriter创建，每次写入数据会同时写入到这一组Writer
type multiWriter struct {
writers []Writer
}
方法
func (t *multiWriter) Write(p []byte) (n int, err error) {
//循环写入
for _, w := range t.writers {
n, err = w.Write(p)
//某个报错则返回
if err != nil {
return
}
//必须将p都写入
if n != len(p) {
err = ErrShortWrite
return
}
}
return len(p), nil
}
//写字符串
func (t *multiWriter) WriteString(s string) (n int, err error) {
var p []byte // lazily initialized if/when needed
for _, w := range t.writers {
//判断是否实现了stringWriter接口
if sw, ok := w.(stringWriter); ok {
n, err = sw.WriteString(s)
} else {
//将string转为slice
if p == nil {
p = []byte(s)
}
//调用Write
n, err = w.Write(p)
}
if err != nil {
return
}
//所有Writer必须都写完
if n != len(s) {
err = ErrShortWrite
return
}
}
return len(s), nil
}
pipe
定义
PipeReader和PipeWriter的底层实现
type pipe struct {
rl sync.Mutex // gates readers one at a time
wl sync.Mutex // gates writers one at a time
l sync.Mutex // protects remaining fields
data []byte // data remaining in pending write
rwait sync.Cond // waiting reader
wwait sync.Cond // waiting writer
rerr error // if reader closed, error to give writes
werr error // if writer closed, error to give reads
}
方法
func (p *pipe) read(b []byte) (n int, err error) {
// One reader at a time.
//上Reader的锁
p.rl.Lock()
defer p.rl.Unlock()
//锁其他字段
p.l.Lock()
defer p.l.Unlock()
//死循环直到有数据可读了
for {
//reader关闭了
if p.rerr != nil {
return 0, ErrClosedPipe
}
//可以读数据了
if p.data != nil {
break
}
//writer关闭了
if p.werr != nil {
return 0, p.werr
}
//阻塞读
p.rwait.Wait()
}
//读数据
n = copy(b, p.data)
p.data = p.data[n:]
//pipe中数据读完了则可以notify Writer
if len(p.data) == 0 {
p.data = nil
p.wwait.Signal()
}
return
}
func (p *pipe) write(b []byte) (n int, err error) {
// pipe uses nil to mean not available
if b == nil {
b = zero[:]
}
// One writer at a time.
p.wl.Lock()
defer p.wl.Unlock()
p.l.Lock()
defer p.l.Unlock()
//Writer关闭抛异常
if p.werr != nil {
err = ErrClosedPipe
return
}
//写入数据
p.data = b
//notify Reader
p.rwait.Signal()
//阻塞，直到Reader读完
for {
if p.data == nil {
break
}
if p.rerr != nil {
err = p.rerr
break
}
if p.werr != nil {
err = ErrClosedPipe
break
}
p.wwait.Wait()
}
n = len(b) - len(p.data)
p.data = nil // in case of rerr or werr
return
}
PipeReader/PipeWriter
定义
通过Pipe()函数创建(*PipeReader, *PipeWriter)
type PipeReader struct {
p *pipe
}
type PipeWriter struct {
p *pipe
}
方法
- Write，调用pipe.write
- Read，调用pipe.read