[译]Go 语言中的流式 IO

作者: 大猫黄 | 来源:发表于2019-04-28 23:44 被阅读3次

原文链接

Streaming IO in Go – Learning the Go Programming Language – Medium

以下为译文

在 Go 中，输入输出操作是通过能读能写的字节流数据模型来实现的。为此，io 包提供了 io.Reader 和 io.Writer 接口来进行输入输出操作，如下所示：

image

Go 附带了许多 API，这些 API 支持来自内存结构，文件，网络连接等资源的流式 IO。本文重点介绍如何自定义实现以及使用标准库中的 io.Reader 和 io.Writer接口创建能够传输流式数据的 Go 程序

io.Reader

由 io.Reader 接口表示的读取器将数据从某些源读取到缓冲区，可以像用水管输送水流一样来传送它，如下所示

image

对于要用作读取器的类型，它必须从接口 io.Reader 实现 Read(p [] byte)方法，如下所示：

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

Read() 方法的实现应返回读取的字节数或发生的错误。如果数据源已输出全部内容，则 Read 应返回 io.EOF

读取规则(补充)

在 Reddit 反馈之后，我决定添加有关读取规则的这一部分。读取器的行为取决于它的实现，但是你应该知道从读取器读取数据时， io.Reader 中的一些规则：

译者注：p 为缓冲区，n 为字节数

如果可能，Read() 将读取 len(p) 到 p
调用 Read() 后，返回的字节数 n 可能小于 len(p)
出错时，Read() 仍可在缓冲区 p 中返回 n 个字节。例如，从突然关闭的 TCP 套接字读取。取决于您的程序设计，您可以选择将字节保存在 p 中或重新尝试从 TCP 套接字中读取
当 Read() 读完所有可用数据时，读取器可能返回非零 n 和 err = io.EOF。尽管如此，您可以自己实现返回规则，如可以选择在流的末尾返回非零 n 和 err = nil。在这种情况下，任何后续读取必须返回 n = 0，err = io.EOF
最后，调用 Read() 返回 n = 0 和 err = nil 并不意味着 EOF，因为下一次调用 Read() 可能会返回更多数据

如您所见，直接从读取器读取流数据可能会非常棘手。幸运的是，标准库中的读取器使用的一些方法使其易于流式传输。不过，在使用读取器之前，请查阅其文档

从读取器中流式传输数据

直接从读取器流式传输数据很容易。Read 方法被设计为在循环内调用，每次迭代时，它从源读取一大块数据并将其放入缓冲区 p 中。直到 Read 方法返回io.EOF 错误

以下是一个简单的示例，它使用 string.NewReader(string) 创建的字符串读取器来从字符串源中流式传输字节值:

func main() {
    reader := strings.NewReader("Clear is better than clever")
    p := make([]byte, 4)
    for {
        n, err := reader.Read(p)
        if err == io.EOF {
            break
        }
        fmt.Println(string(p[:n]))
    }
}

上面的源代码用 make([] byte，4) 创建一个 4 字节长的传输缓冲区 p。缓冲区故意保持小于字符串源的长度, 这是为了演示如何从大于缓冲区的源正确传输数据块

更新: Reddit 上有人指出上面的代码中有 bug, 它永远不会捕获非零错误 err != io.EOF . 以下修复了代码:

func main() {
    reader := strings.NewReader("Clear is better than clever")
    p := make([]byte, 4)

    for {
        n, err := reader.Read(p)
        if err != nil{
            if err == io.EOF {
            fmt.Println(string(p[:n])) //should handle any remainding bytes.
            break
            }
            fmt.Println(err)
            os.Exit(1)
        }
        fmt.Println(string(p[:n]))
    }
}

自定义一个 io.Reader

上一节使用标准库中的现有 IO 读取器实现。现在，让我们看看如何编写自己的读取器。以下是 io.Reader 的简单实现，它从流中过滤掉非字母字符。

package main

import (
    "fmt"
    "io"
)

// alphaReader is a simple implementation of an io.Reader
// that streams only alpha chars from its string source.
type alphaReader struct {
    src string
    cur int
}

func newAlphaReader(src string) *alphaReader {
    return &alphaReader{src: src}
}

func alpha(r byte) byte {
    if (r >= 'A' && r <= 'Z') || (r >= 'a' && r <= 'z') {
        return r
    }
    return 0
}

func (a *alphaReader) Read(p []byte) (int, error) {
    if a.cur >= len(a.src) {
        return 0, io.EOF
    }

    x := len(a.src) - a.cur
    n, bound := 0, 0
    if x >= len(p) {
        bound = len(p)
    } else if x <= len(p) {
        bound = x
    }

    buf := make([]byte, bound)
    for n < bound {
        if char := alpha(a.src[a.cur]); char != 0 {
            buf[n] = char
        }
        n++
        a.cur++
    }
    copy(p, buf)
    return n, nil
}

func main() {
    reader := newAlphaReader("Hello! It's 9am, where is the sun?")
    p := make([]byte, 4)
    for {
        n, err := reader.Read(p)
        if err == io.EOF {
            break
        }
        fmt.Print(string(p[:n]))
    }
    // or use io.Copy
    // io.Copy(os.Stdout, reader)
    fmt.Println()
}

程序执行时，输出:

$> go run alpha_reader.go
HelloItsamwhereisthesun

链式读取器

标准库已经实现了许多读取器。使用读取器作为另一个读取器的源是一种常见的习语。读取器的这种链接允许一个读取器重用另一个读取器的逻辑，就像在下面的源代码片段中所做的那样，更新 alphaReader 以接受 io.Reader 作为其源。这通过将流管理问题推向根读取器来降低代码的复杂性。

package main

import (
    "fmt"
    "io"
    "strings"
)

// alphaReader is a simple implementation of an io.Reader
// that streams only alpha chars from its string source.
// This example uses another reader as data source.
type alphaReader struct {
    reader io.Reader
}

func newAlphaReader(reader io.Reader) *alphaReader {
    return &alphaReader{reader: reader}
}

func alpha(r byte) byte {
    if (r >= 'A' && r <= 'Z') || (r >= 'a' && r <= 'z') {
        return r
    }
    return 0
}

func (a *alphaReader) Read(p []byte) (int, error) {
    n, err := a.reader.Read(p)
    if err != nil {
        return n, err
    }
    buf := make([]byte, n)
    for i := 0; i < n; i++ {
        if char := alpha(p[i]); char != 0 {
            buf[i] = char
        }
    }

    copy(p, buf)
    return n, nil
}

func main() {
    // use an io.Reader as source for alphaReader
    reader := newAlphaReader(strings.NewReader("Hello! It's 9am, where is the sun?"))
    p := make([]byte, 4)
    for {
        n, err := reader.Read(p)
        if err == io.EOF {
            break
        }
        fmt.Print(string(p[:n]))
    }
    fmt.Println()
}

这种方法的另一个优点是 alphaReader 现在能够从任何读取器实现中读取。例如，以下代码段显示了如何将 alphaReader 与 os.File 源结合以过滤掉文件中的非字母字符：

package main

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

// alphaReader is a simple implementation of an io.Reader
// that streams only alpha chars from its string source.
// This example uses another reader as data source.
type alphaReader struct {
    reader io.Reader
}

func newAlphaReader(reader io.Reader) *alphaReader {
    return &alphaReader{reader: reader}
}

func alpha(r byte) byte {
    if (r >= 'A' && r <= 'Z') || (r >= 'a' && r <= 'z') {
        return r
    }
    return 0
}

func (a *alphaReader) Read(p []byte) (int, error) {
    n, err := a.reader.Read(p)
    if err != nil {
        return n, err
    }
    buf := make([]byte, n)
    for i := 0; i < n; i++ {
        if char := alpha(p[i]); char != 0 {
            buf[i] = char
        }
    }

    copy(p, buf)
    return n, nil
}

func main() {
    // use an io.Reader as source for alphaReader
    file, err := os.Open("./alpha_reader2.go")
    if err != nil {
        fmt.Println(err)
        os.Exit(1)
    }
    defer file.Close()

    reader := newAlphaReader(file)
    p := make([]byte, 4)
    for {
        n, err := reader.Read(p)
        if err == io.EOF {
            break
        }
        fmt.Print(string(p[:n]))
    }
    fmt.Println()
}

io.Writer

由接口 io.Writer 表示的写入器从缓冲区流式传输数据并将其写入目标资源，如下所示

image

所有流写入器必须从接口 io.Writer 实现方法 Write(p [] byte)。该方法旨在从缓冲区 p 读取数据并将其写入指定的目标资源

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

Write() 方法的实现应返回写入的字节数或发生的错误

使用写入器

标准库附带了许多预先实现的 io.Writer 类型。直接使用写入器很简单，如下面的代码片段所示，它使用 bytes.Buffer 作为 io.Writer 将数据写入内存缓冲区

package main

import (
    "bytes"
    "fmt"
    "os"
)

func main() {
    proverbs := []string{
        "Channels orchestrate mutexes serialize",
        "Cgo is not Go",
        "Errors are values",
        "Don't panic",
    }
    var writer bytes.Buffer

    for _, p := range proverbs {
        n, err := writer.Write([]byte(p))
        if err != nil {
            fmt.Println(err)
            os.Exit(1)
        }
        if n != len(p) {
            fmt.Println("failed to write data")
            os.Exit(1)
        }
    }

    fmt.Println(writer.String())
}

自定义一个 io.Writer

本节中的代码显示了如何实现一个名为 chanWriter 的自定义 io.Writer，它将其内容作为字节序列写入 Go 通道。

package main

import "fmt"

type chanWriter struct {
    ch chan byte
}

func newChanWriter() *chanWriter {
    return &chanWriter{make(chan byte, 1024)}
}

func (w *chanWriter) Chan() <-chan byte {
    return w.ch
}

func (w *chanWriter) Write(p []byte) (int, error) {
    n := 0
    for _, b := range p {
        w.ch <- b
        n++
    }
    return n, nil
}

func (w *chanWriter) Close() error {
    close(w.ch)
    return nil
}

func main() {
    writer := newChanWriter()
    go func() {
        defer writer.Close()
        writer.Write([]byte("Stream "))
        writer.Write([]byte("me!"))
    }()
    for c := range writer.Chan() {
        fmt.Printf("%c", c)
    }
    fmt.Println()
}

要使用写入器，代码只需在函数 main() 中调用方法 writer.Write() (在单独的goroutine 中）。因为 chanWriter 还实现了接口 io.Closer，所以调用方法writer.Close() 来正确关闭通道，以避免在访问通道时出现死锁

Useful types and packages for IO

如前所述，Go 标准库附带了许多有用的功能和其他类型，可以轻松使用流式IO

os.File

os.File 类型表示本地系统上的文件。它实现了 io.Reader 和 io.Writer，因此可以在任何流 IO 上下文中使用。例如，以下示例显示如何将连续的字符串切片直接写入文件

package main

import (
    "fmt"
    "os"
)

func main() {
    proverbs := []string{
        "Channels orchestrate mutexes serialize\n",
        "Cgo is not Go\n",
        "Errors are values\n",
        "Don't panic\n",
    }
    file, err := os.Create("./proverbs.txt")
    if err != nil {
        fmt.Println(err)
        os.Exit(1)
    }
    defer file.Close()

    for _, p := range proverbs {
        n, err := file.Write([]byte(p))
        if err != nil {
            fmt.Println(err)
            os.Exit(1)
        }
        if n != len(p) {
            fmt.Println("failed to write data")
            os.Exit(1)
        }
    }
    fmt.Println("file write done")
}

相反，io.File 类型可以用作读取器来从本地文件系统流式传输文件的内容。例如，以下源代码段读取文件并打印其内容：

package main

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

func main() {
    file, err := os.Open("./proverbs.txt")
    if err != nil {
        fmt.Println(err)
        os.Exit(1)
    }
    defer file.Close()

    p := make([]byte, 4)
    for {
        n, err := file.Read(p)
        if err == io.EOF {
            break
        }
        fmt.Print(string(p[:n]))
    }
}

Standard output, input, and error

os 包公开三个变量，os.Stdout，os.Stdin 和 os.Stderr，它们的类型为* os.File，分别表示操作系统标准输出\输入\错误的文件句柄。例如，以下源代码段直接打印到标准输出：

package main

import (
    "fmt"
    "os"
)

func main() {
    proverbs := []string{
        "Channels orchestrate mutexes serialize\n",
        "Cgo is not Go\n",
        "Errors are values\n",
        "Don't panic\n",
    }

    for _, p := range proverbs {
        n, err := os.Stdout.Write([]byte(p))
        if err != nil {
            fmt.Println(err)
            os.Exit(1)
        }
        if n != len(p) {
            fmt.Println("failed to write data")
            os.Exit(1)
        }
    }
}

io.Copy()

io.Copy() 方法可以轻松地将数据从源读取器传输到目标写入器。它抽象出 for 循环模式（我们到目前为止已经看到）并正确处理 io.EOF 和字节计数。

以下显示了以前程序的简化版本，该程序复制内存读取器 proberbs 的内容并将其复制到 writer 文件：

package main

import (
    "bytes"
    "fmt"
    "io"
    "os"
)

func main() {
    proverbs := new(bytes.Buffer)
    proverbs.WriteString("Channels orchestrate mutexes serialize\n")
    proverbs.WriteString("Cgo is not Go\n")
    proverbs.WriteString("Errors are values\n")
    proverbs.WriteString("Don't panic\n")

    file, err := os.Create("./proverbs.txt")
    if err != nil {
        fmt.Println(err)
        os.Exit(1)
    }
    defer file.Close()

    // copy from reader data into writer file
    if _, err := io.Copy(file, proverbs); err != nil {
        fmt.Println(err)
        os.Exit(1)
    }
    fmt.Println("file created")
}

同样，我们可以使用 io.Copy() 函数重写以前从文件读取并打印到标准输出的程序，如下所示

package main

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

func main() {
    file, err := os.Open("./proverbs.txt")
    if err != nil {
        fmt.Println(err)
        os.Exit(1)
    }
    defer file.Close()

    if _, err := io.Copy(os.Stdout, file); err != nil {
        fmt.Println(err)
        os.Exit(1)
    }
}

io.WriterString()

此函数提供了将字符串值写入指定写入器的便利，如下所示

package main

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

func main() {
    file, err := os.Create("./magic_msg.txt")
    if err != nil {
        fmt.Println(err)
        os.Exit(1)
    }
    defer file.Close()
    if _, err := io.WriteString(file, "Go is fun!"); err != nil {
        fmt.Println(err)
        os.Exit(1)
    }
}

Pipe writers and readers

io.PipeWriter 类型和 io.PipeReader 模型 IO 操作在内存管道中。数据被写入管道的 writer-end，并使用单独的 go 例程在管道的 reader-end 上读取。下面使用 io.Pipe() 创建管道读取器/写入器对，然后使用 io.Pipe() 将数据从缓冲区 proverbs 复制到 io.Stdout, 如下所示

package main

import (
    "bytes"
    "io"
    "os"
)

func main() {
    proverbs := new(bytes.Buffer)
    proverbs.WriteString("Channels orchestrate mutexes serialize\n")
    proverbs.WriteString("Cgo is not Go\n")
    proverbs.WriteString("Errors are values\n")
    proverbs.WriteString("Don't panic\n")

    piper, pipew := io.Pipe()

    // write in writer end of pipe
    go func() {
        defer pipew.Close()
        io.Copy(pipew, proverbs)
    }()

    // read from reader end of pipe.
    io.Copy(os.Stdout, piper)
    piper.Close()
}

Buffered IO

Go 通过 bufio 包支持缓冲 IO，可以轻松处理文本内容。例如，以下程序逐行读取文件以值 '\ n' 分隔的内容

package main

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

func main() {
    file, err := os.Open("./planets.txt")
    if err != nil {
        fmt.Println(err)
        os.Exit(1)
    }
    defer file.Close()
    reader := bufio.NewReader(file)

    for {
        line, err := reader.ReadString('\n')
        if err != nil {
            if err == io.EOF {
                break
            } else {
                fmt.Println(err)
                os.Exit(1)
            }
        }
        fmt.Print(line)
    }

}

Util package

ioutil 包为 IO 提供了几个便利功能。例如，以下使用函数 ReadFile 将文件内容加载到[]字节中

package main

import (
    "fmt"
    "io/ioutil"
    "os"
)

func main() {
    bytes, err := ioutil.ReadFile("./planets.txt")
    if err != nil {
        fmt.Println(err)
        os.Exit(1)
    }
    fmt.Printf("%s", bytes)
}

结论

本文介绍如何使用 io.Reader 和 io.Writer 接口在程序中实现流式 IO。阅读本文后，您应该能够了解如何创建使用 io 包流式传输 IO 数据的程序，有很多示例向您展示了如何为自定义功能创建自己的 io.Reader 和 io.Writer 类型。

这是一个介绍性的讨论，几乎没有涉及支持流 IO 的 Go 包的范围。例如，我们没有进入文件 IO，缓冲 IO，网络 IO或格式化 IO（为将来的写作而保留）。我希望这能让你了解 Go 的流式 IO 惯用语是什么

[译]Go 语言中的流式 IO

原文链接

以下为译文

io.Reader

读取规则(补充)

从读取器中流式传输数据

自定义一个 io.Reader

链式读取器

io.Writer

使用写入器

自定义一个 io.Writer

Useful types and packages for IO

os.File

Standard output, input, and error

io.Copy()

io.WriterString()

Pipe writers and readers

Buffered IO

Util package

结论

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

深入浅出golang

Golang