读取超时。

Question

我想做一些类似 unix 的东西tail -f，但是在通过 Go 的Cmd设施运行的进程产生的输出上。显然，我的 google-fu 没有达到标准，但我确实找到了这篇文章，它引导我编写以下代码，几乎可以正常工作，但我希望能得到帮助。如果重要的话，我会在 Mac 上运行它。首先，这是编译为slowroll可执行文件的最小程序：package mainimport (    "fmt"    "time")func main() {    line := 1    for {        fmt.Println("This is line", line)        line += 1        time.Sleep(2 * time.Second)    }}运行时，它会产生以下输出，每 2 秒一行：    > ./slowroll    This is line 1    This is line 2    This is line 3    This is line 4等等。这是尝试读取此内容的包代码，但允许超时以便可以完成其他操作：package timeout_ioimport (    "bufio"    "bytes"    "context"    "errors"    "time")const BufferSize = 4096var ErrTimeout = errors.New("timeout")type TimeoutReader struct {    b *bufio.Reader    t time.Duration}func NewTimeoutReader(stdOut *bytes.Buffer) *TimeoutReader {    return &TimeoutReader{b: bufio.NewReaderSize(stdOut, BufferSize), t: 0}}func (r *TimeoutReader) SetTimeout(t time.Duration) time.Duration {    prev := r.t    r.t = t    return prev}type CallResponse struct {    Resp string    Err  error}func helper(r *bufio.Reader) <-chan *CallResponse {    respChan := make(chan *CallResponse, 1)    go func() {        resp, err := r.ReadString('\n')        if err != nil {            respChan <- &CallResponse{resp, err}        } else {            respChan <- &CallResponse{resp, nil}        }        return    }()    return respChan}func (r *TimeoutReader) ReadLineCtx(ctx context.Context) (string, error) {    select {    case <-ctx.Done():        return "", ErrTimeout    case respChan := <-helper(r.b):        return respChan.Resp, respChan.Err    }}

Answer 1

通过创建管道并从该管道读取来简化代码：

cmd := exec.Command("./slowroll")

stdout, _ := cmd.StdoutPipe()

if err := cmd.Start(); err != nil {

    log.Fatal(err)

}

s := bufio.NewScanner(stdout)

for s.Scan() {

    fmt.Printf("%s\n", s.Bytes())

}

如果您的目标是监视 stderr 和 stdin 的组合输出，则对两者使用相同的管道：

cmd := exec.Command("./slowroll")

combined, _ := cmd.StdoutPipe()

cmd.Stderr = cmd.Stdout // <-- use stdout pipe for stderr

if err := cmd.Start(); err != nil {

    log.Fatal(err)

}

s := bufio.NewScanner(combined)

for s.Scan() {

    fmt.Printf("%s\n", s.Bytes())

}

问题中的代码在 stdOut bytes.Buffer 上有数据竞争。

Answer 2

if err != timeout_io.ErrTimeout && err != io.EOF { ...; break; }

在这样的条件下， anErrTimeout将被默默地忽略并且不会中断您的阅读循环。

另请注意，到达io.EOF会在无限循环中发送您的程序（尝试使用echo "Hello"而不是./slowroll作为命令）。

您可能希望将break指令放在if 块之后：

if err != timeout_io.ErrTimeout && err != io.EOF {

    fmt.Println("ReadLine got error", err)

}

break

Answer 3

昨晚深夜意识到我有点像围棋的标准行为。

应该解释说目标是能够同时观看标准输出和标准错误。

接受上面@Zombo 的建议，我切换到cmd.StdoutPipeand cmd.StderrPipe。

主要思想是只有 goroutines 读取管道并将找到的内容放入通道，然后select在通道之间。

所以slowroll.go不会产生无限输出，以表明 EOF 不会导致无限循环：

package main

import (

    "fmt"

    "os"

    "time"

)

func main() {

    line := 1

    for {

        fmt.Println("This is line", line)

        line += 1

        time.Sleep(2 * time.Second)

        if line%3 == 0 {

            fmt.Fprintf(os.Stderr, "This is error %d\n", line)

        }

        if line > 10 {

            break

        }

    }

}

现在更简单的工作watcher.go是：

package main

import (

    "bufio"

    "fmt"

    "os"

    "os/exec"

    "sync"

)

func main() {

    runCommand := &exec.Cmd{

        Path: "./slowroll",

    }

    stdOut, err := runCommand.StdoutPipe()

    if err != nil {

        fmt.Println("Can't create StdoutPipe:", err)

        os.Exit(1)

    }

    stdErr, err := runCommand.StderrPipe()

    if err != nil {

        fmt.Println("Can't create StderrPipe:", err)

        os.Exit(1)

    }

    var wg sync.WaitGroup

    go func(wg *sync.WaitGroup) {

        defer wg.Done()

        err := runCommand.Run()

        if err != nil {

            fmt.Println("failed due to error:", err)

            os.Exit(1)

        }

    }(&wg)

    wg.Add(1)

    stdOutChan := make(chan string, 1)

    go func(wg *sync.WaitGroup) {

        defer wg.Done()

        scanner := bufio.NewScanner(stdOut)

        for scanner.Scan() {

            stdOutChan <- string(scanner.Bytes())

        }

        fmt.Println("Ran out of stdout input, read thread bailing.")

        close(stdOutChan)

    }(&wg)

    wg.Add(1)

    stdErrChan := make(chan string, 1)

    go func(wg *sync.WaitGroup) {

        defer wg.Done()

        scanner := bufio.NewScanner(stdErr)

        for scanner.Scan() {

            stdErrChan <- string(scanner.Bytes())

        }

        fmt.Println("Ran out of stderr input, read thread bailing.")

        close(stdErrChan)

    }(&wg)

    wg.Add(1)

    index := 1

    keepGoing := true

    for keepGoing {

        select {

        case res, isOpen := <-stdOutChan:

            if !isOpen {

                fmt.Println("stdOutChan is no longer open, main bailing.")

                keepGoing = false

            } else {

                fmt.Println(index, "s:", res)

                index += 1

            }

        case res, isOpen := <-stdErrChan:

            if !isOpen {

                fmt.Println("stdErrChan is no longer open, main bailing.")

                keepGoing = false

            } else {

                fmt.Println(index, "error s:", res)

                index += 1

            }

        }

    }

    wg.Wait()

    fmt.Println("Done!")

}

输出：

    > go run watcher.go

    1 s: This is line 1

    2 s: This is line 2

    3 error s: This is error 3

    4 s: This is line 3

    5 s: This is line 4

    6 s: This is line 5

    7 s: This is line 6

    8 error s: This is error 6

    9 s: This is line 7

    10 s: This is line 8

    11 s: This is line 9

    12 error s: This is error 9

    13 s: This is line 10

    Ran out of stdout input, read thread bailing.

    stdOutChan is no longer open, main bailing.

    Ran out of stderr input, read thread bailing.

    Done!

显然，可以进行一些重构，但它可以工作，这就是目标。