golang 中通过strings/bytes/bufio 等包实现相关IO

Posted 2023-04-25 终点就在前方

在 go 的 IO 中，除了 io 、os 等包，我们还有 strings 、 bytes 、bufio 等实现 IO 读写，这其中有些实现了 io.Reader，有些同时实现了 io.Reader和io.Writer 接口。接下来我们一个个看相关的使用。

1.strings

在 strings包 中，只实现了 Reader，我们看看其代码：

// 实现的 reader 
type Reader struct 
	s        string
	i        int64 // current reading index
	prevRune int   // index of previous rune; or < 0


// 构造函数
// NewReader returns a new Reader reading from s.
// It is similar to bytes.NewBufferString but more efficient and read-only.
func NewReader(s string) *Reader  return &Readers, 0, -1 

// Read implements the io.Reader interface.
func (r *Reader) Read(b []byte) (n int, err error) 
	if r.i >= int64(len(r.s)) 
		return 0, io.EOF
	
	r.prevRune = -1
	n = copy(b, r.s[r.i:])
	r.i += int64(n)
	return

我们通过 NewReader(string) 构建一个 Reader 对象，随后就可以通过 Read() 读取 Reader 的内容，以下是其使用：

package main

import (
	"fmt"
	"strings"
)

func main()  
	s := "Today is monday, what a great start!"
	sr := strings.NewReader(s)

	// read
	cap_sr := sr.Len()
	data := make([]byte, cap_sr)
	n, err := sr.Read(data)
	if err != nil 
		fmt.Printf("read to data failed, err: %s\\n", err)
		return
	
	fmt.Printf("read data size: %d, content: %s\\n", n, data)

main函数结果

read data size: 36, content: Today is monday, what a great start!

2.bytes

在 bytes包 中，内部实现了 io.Reader，分别是 bytes.Reader 和 bytes.Buffer，顾名思义，带 buffer 的就是有一定缓冲，但主要针对的还是 字节IO，下面看看其源码：

bytes.Reader

type Reader struct 
	s        []byte
	i        int64 // current reading index
	prevRune int   // index of previous rune; or < 0


// NewReader returns a new Reader reading from b.
func NewReader(b []byte) *Reader  return &Readerb, 0, -1 

// Read implements the io.Reader interface.
func (r *Reader) Read(b []byte) (n int, err error) 
	if r.i >= int64(len(r.s)) 
		return 0, io.EOF
	
	r.prevRune = -1
	n = copy(b, r.s[r.i:])
	r.i += int64(n)
	return

其实我们也可以看到， bytes 和 strings 的 Reader 很像，只不过一个是字符串，一个是字节切片。

bytes.Buffer

// A Buffer is a variable-sized buffer of bytes with Read and Write methods.
// The zero value for Buffer is an empty buffer ready to use.
type Buffer struct 
	buf      []byte // contents are the bytes buf[off : len(buf)]
	off      int    // read at &buf[off], write at &buf[len(buf)]
	lastRead readOp // last read operation, so that Unread* can work correctly.


func NewBufferString(s string) *Buffer 
	return &Bufferbuf: []byte(s)


func NewBuffer(buf []byte) *Buffer  return &Bufferbuf: buf 


func (b *Buffer) Read(p []byte) (n int, err error) 
	b.lastRead = opInvalid
	if b.empty() 
		// Buffer is empty, reset to recover space.
		b.Reset()
		if len(p) == 0 
			return 0, nil
		
		return 0, io.EOF
	
	n = copy(p, b.buf[b.off:])
	b.off += n
	if n > 0 
		b.lastRead = opRead
	
	return n, nil



func (b *Buffer) Write(p []byte) (n int, err error) 
	b.lastRead = opInvalid
	m, ok := b.tryGrowByReslice(len(p))
	if !ok 
		m = b.grow(len(p))
	
	return copy(b.buf[m:], p), nil

可以看到，在这部分，提供了两个构造方法，通过字节切片或者字符串都可，对于 Read 方法其实和其他的都类似，主要是 Write 中，会将传入的数据追加给 Buffer.buf，所以再次读取内容时，可以看到是读到了新加的内容的。

使用示例

package main

import (
	"bytes"
	"fmt"
)

func main()  
	// bytes.Reader
	fmt.Println("bytes.Reader test...")
	nr := bytes.NewReader([]byte("hello"))

	data := make([]byte, nr.Len())
	n, err := nr.Read(data)
	if err != nil 
		fmt.Printf("read to data failed, err: %s\\n", err)
		return
	
	fmt.Printf("read data1 size: %d, content: %s\\n", n, data)

	// bytes.Buffer
	fmt.Println("bytes.Buffer test...")
	nbs := bytes.NewBufferString("world")
	nbs.Write([]byte(" lalalala..."))

	data2 := make([]byte, nbs.Len())
	n, err = nbs.Read(data2)
	if err != nil 
		fmt.Printf("read to data failed, err: %s\\n", err)
		return
	
	fmt.Printf("read data2 size: %d, content: %s\\n", n, data2)

测试结果

bytes.Reader test...
read data1 size: 5, content: hello
bytes.Buffer test...
read data2 size: 17, content: world lalalala...

3.bufio

相较于前两者，在 bufio 中，实现的结构体就丰富很多，Reader、Writer、ReadWriter(结构体继承前2者) 都有实现，而且我们也可以通过 Writer 写入文件中，实现向文件中写入数据。

源码-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


// NewReader returns a new Reader whose buffer has the default size.
func NewReader(rd io.Reader) *Reader 
	return NewReaderSize(rd, defaultBufSize)


// Read reads data into p.
// It returns the number of bytes read into p.
// The bytes are taken from at most one Read on the underlying Reader,
// hence n may be less than len(p).
// To read exactly len(p) bytes, use io.ReadFull(b, p).
// If the underlying Reader can return a non-zero count with io.EOF,
// then this Read method can do so as well; see the [io.Reader] docs.
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()
	
	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)
			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.
		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 += n
	

	// copy as much as we can
	// Note: if the slice panics here, it is probably because
	// the underlying reader returned a bad count. See issue 49795.
	n = copy(p, b.buf[b.r:b.w])
	b.r += n
	b.lastByte = int(b.buf[b.r-1])
	b.lastRuneSize = -1
	return n, nil


// 按行读取内容
func (b *Reader) ReadLine() (line []byte, isPrefix bool, err error) 
	line, err = b.ReadSlice(\'\\n\')
	if err == ErrBufferFull 
		// Handle the case where "\\r\\n" straddles the buffer.
		if len(line) > 0 && line[len(line)-1] == \'\\r\' 
			// Put the \'\\r\' back on buf and drop it from line.
			// Let the next call to ReadLine check for "\\r\\n".
			if b.r == 0 
				// should be unreachable
				panic("bufio: tried to rewind past start of buffer")
			
			b.r--
			line = line[:len(line)-1]
		
		return line, true, nil
	

	if len(line) == 0 
		if err != nil 
			line = nil
		
		return
	
	err = nil

	if line[len(line)-1] == \'\\n\' 
		drop := 1
		if len(line) > 1 && line[len(line)-2] == \'\\r\' 
			drop = 2
		
		line = line[:len(line)-drop]
	
	return

源码-Writer

// Writer implements buffering for an io.Writer object.
// If an error occurs writing to a Writer, no more data will be
// accepted and all subsequent writes, and Flush, will return the error.
// After all data has been written, the client should call the
// Flush method to guarantee all data has been forwarded to
// the underlying io.Writer.
type Writer struct 
	err error
	buf []byte
	n   int
	wr  io.Writer


// NewWriter returns a new Writer whose buffer has the default size.
// If the argument io.Writer is already a Writer with large enough buffer size,
// it returns the underlying Writer.
func NewWriter(w io.Writer) *Writer 
	return NewWriterSize(w, defaultBufSize)


// Write writes the contents of p into the buffer.
// It returns the number of bytes written.
// If nn < len(p), it also returns an error explaining
// why the write is short.
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
			b.Flush()
		
		nn += n
		p = p[n:]
	
	if b.err != nil 
		return nn, b.err
	
	n := copy(b.buf[b.n:], p)
	b.n += n
	nn += n
	return nn, nil


// Flush writes any buffered data to the underlying io.Writer.
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

这里需要注意的是，在 Writer 中调用完 Write 方法后，应该 Flush() 操作，才能将实际将数据转到 Writer.wr 中。

使用示例

package main

import (
	"bufio"
	"fmt"
	"io"
	"os"
)

func main()  
	// read
	var filename string = "D:\\\\demo1\\\\src\\\\demo23\\\\go-io\\\\file\\\\file.txt"
	f, _ := os.Open(filename)
	defer f.Close()
	nr := bufio.NewReader(f)
	for 
		line, _, err := nr.ReadLine()
		fmt.Println(string(line))
		if err == io.EOF 
			break
		
	

	// write
	fw, _ := os.Create("bufio_w.txt")
	defer fw.Close()
	nw := bufio.NewWriter(fw)
	nw.WriteString("bufio write:\\n")
	nw.WriteString("hello\\n")
	nw.WriteString("world\\n")
	nw.Flush()

测试结果

静夜思 - 唐·李白
床前看月光，疑是地上霜。
举头望山月，低头思故乡。

bufi_w.txt

bufio write:
hello
world

以上就是对 go 内置的各种 io 的总结，接下来将完善网络IO部分。

TSINGSEE青犀视频编译中通过GoLang代码修改Linux服务的ulimit的实现

在EasyDSS、EasyGBS、EasyCVR和EasyNVR等产品中，如果部署在Linux系统下，当接入设备过多时，由于操作系统对于打开文件数有一定限制，会出现程序报错的现象，出现 “open too many files”错误。这个错误我们之前也介绍过一些解决方法，但是鉴于很多用户都有类似情况，我们还是决定对代码进行研修，消除此类报错。

这里我们需要手动修改操作系统的配置，然后重启操作系统使操作设置正确。但是手动操作比较麻烦，因此直接使用代码实现该功能。代码如下：

// Package base Copyright 2021 TSINGSEE.
// http://www.tsingsee.com
// ulimit 服务限制
// Creat By Sam
// History (Name, Time, Desc)
// (Sam, 20211202, 创建文件)
package main

import (
   "bufio"
   "bytes"
   "errors"
   "fmt"
   "io"
   "io/ioutil"
   "os/exec"
   "runtime"
   "strings"
)

var ErrOsNotSupport = errors.New("OS not support, only linux can use this function")

func main() 
   fmt.Println("修改 ulimit 配置", ModifyServiceUlimit(1000000))
   fmt.Println("设置 ulimit 有效", SystemdConfEffect())


// 修改服务的 ulimit
func ModifyServiceUlimit(ulimit int) error 
   if runtime.GOOS != "linux" 
      return ErrOsNotSupport
   

   //confFile := "C:\\\\Users\\\\sam\\\\Desktop\\\\going\\\\system.conf"
   confFile := "/etc/systemd/system.conf"
   readBytes, err := ioutil.ReadFile(confFile)
   if err != nil 
      return err
   

   oriContent := string(readBytes)
   strReader := strings.NewReader(oriContent)
   br := bufio.NewReader(strReader)
   newLine := fmt.Sprintf("DefaultLimitNOFILE=%d\\n", ulimit)
   findNofile := false
   var buffer bytes.Buffer

   for 
      l, e := br.ReadBytes('\\n')
      if e == io.EOF 
         break
      
      line := string(l)
      if strings.Contains(line, "DefaultLimitNOFILE") 
         line = newLine
         findNofile = true
      
      buffer.WriteString(line)
   

   // 如果未找到，则写入 newLine
   if !findNofile 
      buffer.WriteString(newLine)
   

   //fmt.Println(buffer.String())
   ioutil.WriteFile(confFile, buffer.Bytes(), 666)

   return nil


// systemctl daemon-reexec
// 重启 systemd 配置，使 conf 生效
func SystemdConfEffect() error 
   if runtime.GOOS != "linux" 
      return ErrOsNotSupport
   
   cmd := exec.Command("systemctl", "daemon-reexec")
   return cmd.Run()

运行以上代码，将对应的打开文件限制修改为1000000。重启运行的服务，查看对应服务的打开文件数为1000000即可，如下：