协调多个 goroutine 的错误状态？

Question

我有三个相互依赖的 goroutine，我正在通过通道协调它们。其中两个例程是写入程序，一个是命名管道的读取程序。

但是，每个例程都可能有错误。例如，

os.OpenFile

可能会在其中几个例程中失败。如果发生这种情况，我们要立即退出例程。但是，如果我们这样做，那么其他例程将被阻塞，因为通道永远不会被消耗。

如何协调和处理这三个例程的错误状态？我是否需要为每个操作提供一个错误通道，然后在每个其他例程中交叉检查它们？

func main() {
    syscall.Mkfifo("p.pipe", 0666)
    writer, _ := os.Create("output.csv")

    queryDone := make(chan error)
    waitingDone := make(chan error)
    copyDone := make(chan error)
    go func() {
        // Open for reading and copy to http handler
        r, _ := os.OpenFile("p.pipe", os.O_RDONLY, os.ModeNamedPipe)
        n, _ := io.Copy(writer, r)
        e := <-waitingDone
        r.Close()
        copyDone <- e
    }()

    go func() {
        // Open for writing
        pipe, _ := os.OpenFile("p.pipe", os.O_WRONLY|os.O_APPEND, os.ModeNamedPipe)
        e := <-queryDone
        _ = pipe.Close()
        waitingDone <- e
    }()

    go func() {
        // DB query
        db, _ := sql.Open("duckdb", "")
        defer db.Close()
        _, err = db.Exec("COPY (select 1) TO 'p.pipe' (FORMAT CSV)")
        queryDone <- err
    }()

    e := <-copyDone
    fmt.Println("Done", e)
}

Answer 1

使用可取消的上下文，如果发生错误则取消它：

ctx, cancel := context.WithCancel(context.Background())
defer cancel()
// Channel to capture the first error
errCh:=make(chan error,1)

// Template for goroutines
go func() {
   for {
     err:=doSometing()
     if err!=nil {
        // Record the error in the channel
        select {
           case errCh<-err:
           default:
        }
        // cancel all goroutines
        cancel()
        return
     }
     if ctx.Err()!=nil {
         return // Error happened somewhere else
     }
}()
...
// Check if error happened. This will collect the first error
var err error
select {
  case err=<-errCh:
  default:
}
if err!=nil {
   // Handle error
}

还有第三方库可以帮助协调多个 goroutine 之间的错误状态。