conc：更好的go结构化并发

╰+哭是因爲堅強的太久メ 2024-03-29 14:42 149阅读 0赞

## conc：更好的go结构化并发 ##

*本文译自：[conc: better structured concurrency for go][conc_ better structured concurrency for go]*

### 原文截图 ###

![在这里插入图片描述][03331c245fc34fae864b2640d608162a.png_pic_center]

### 译文 ###

`conc`是你在 go 中进行结构化并发的工具带，使常见任务更容易、更安全。

go get github.com/sourcegraph/conc

## 一目了然 ##

*  如果[`conc.WaitGroup`][conc.WaitGroup]您只想要更安全的版本，请使用`sync.WaitGroup`
 *  [`pool.Pool`][pool.Pool]如果你想要一个并发限制的任务运行器，请使用
 *  [`pool.ResultPool`][pool.ResultPool]如果您想要一个收集任务结果的并发任务运行器，请使用
 *  [`pool.(Result)?ErrorPool`][pool._Result_ErrorPool]如果您的任务容易出错，请使用
 *  [`pool.(Result)?ContextPool`][pool._Result_ContextPool]如果您的任务应在失败时取消，则使用
 *  [`stream.Stream`][stream.Stream]如果要与串行回调并行处理有序的任务流，请使用
 *  [`iter.Map`][iter.Map]如果您想同时映射一个切片，请使用
 *  如果[`iter.ForEach`][iter.ForEach]您想同时迭代一个切片，请使用
 *  [`panics.Catcher`][panics.Catcher]如果你想在你自己的 goroutines 中捕获恐慌，请使用

所有池都是用 [`pool.New()`][pool.New] or 创建的[`pool.NewWithResults[T]()`][pool.NewWithResults_T]，然后用方法配置：

*  [`p.WithMaxGoroutines()`][p.WithMaxGoroutines]配置池中 goroutines 的最大数量
 *  [`p.WithErrors()`][p.WithErrors]配置池以运行返回错误的任务
 *  [`p.WithContext(ctx)`][p.WithContext_ctx]将池配置为运行应在第一次出错时取消的任务
 *  [`p.WithFirstError()`][p.WithFirstError]配置错误池以仅保留第一个返回的错误而不是聚合错误
 *  [`p.WithCollectErrored()`][p.WithCollectErrored]配置结果池以收集结果，即使任务出错

## 目标 ##

该软件包的主要目标是：

1.  让泄漏 goroutines 变得更难
2.  从容应对恐慌
3.  使并发代码更易于阅读

### 目标 \#1：让泄漏 goroutines 变得更难 ###

使用 goroutine 时的一个常见痛点是清理它们。很容易发出一条`go`语句而未能正确等待它完成。

`conc`采取固执己见的立场，即所有并发都应在范围内。也就是说，goroutines 应该有一个所有者，并且该所有者应该始终确保其拥有的 goroutines 正确退出。

在`conc`中，goroutine 的所有者始终是`conc.WaitGroup`. Goroutines 在 with 中产生`WaitGroup`，`(*WaitGroup).Go()`并且 `(*WaitGroup).Wait()`应该总是在`WaitGroup`超出范围之前被调用。

在某些情况下，您可能希望生成的 goroutine 比调用者的范围更持久。在这种情况下，您可以将 a 传递给`WaitGroup`生成函数。

func main() {
        
        var wg conc.WaitGroup
        defer wg.Wait()
        startTheThing(&wg)
    }
    func startTheThing(wg *conc.WaitGroup) {
        
        wg.Go(func() {
         ... })
    }

有关为什么作用域并发很好的更多讨论，请查看[此博客文章][Link 1]。

### 目标 2：从容应对恐慌 ###

在长时间运行的应用程序中，goroutines 的一个常见问题是处理恐慌。在没有 panic 处理程序的情况下生成的 goroutine 将在 panic 时使整个进程崩溃。这通常是不可取的。

但是，如果您确实向 goroutine 添加了一个 panic 处理程序，那么一旦您捕获到 panic，您将如何处理它？一些选项：

1.  忽略它
2.  登录
3.  将其转化为错误并将其返回给 goroutine spawner
4.  将 panic 传播到 goroutine spawner

忽略恐慌是一个坏主意，因为恐慌通常意味着实际上存在问题并且应该有人修复它。

仅仅记录 panic 也不是很好，因为这样就没有迹象表明 spawner 发生了不好的事情，并且即使您的程序处于非常糟糕的状态，它也可能会继续正常运行。

(3) 和 (4) 都是合理的选择，但都需要 goroutine 有一个所有者，该所有者可以实际接收到出错的消息。对于使用 spawned 生成的 goroutine 通常不是这样`go`，但在`conc` 包中，所有 goroutine 都有一个所有者，必须收集生成的 goroutine。在 conc 包中，`Wait()`如果任何派生的 goroutines 恐慌，则任何调用都会恐慌。此外，它使用来自子 goroutine 的堆栈跟踪来装饰恐慌值，这样您就不会丢失有关导致恐慌的原因的信息。

每次生成一些东西时都正确地完成这一切`go`并不是微不足道的，它需要大量的样板文件，这使得代码的重要部分更难阅读，所以`conc`这对你来说也是如此。

<table> 
 <tbody> 
  <tr> 
   <th><code>stdlib</code></th> 
   <th><code>conc</code></th> 
  </tr> 
  <tr> 
   <td> </td> 
  </tr> 
 </tbody> 
</table>

type caughtPanicError struct {
        
        val   any
        stack []byte
    }
    func (e *caughtPanicError) Error() string {
        
        return fmt.Sprintf(
            "panic: %q\n%s",
            e.val,
            string(e.stack)
        )
    }
    func main() {
        
        done := make(chan error)
        go func() {
        
            defer func() {
        
                if v := recover(); v != nil {
        
                    done <- caughtPanicError{
        
                        val: v,
                        stack: debug.Stack()
                    }
                } else {
        
                    done <- nil
                }
            }()
            doSomethingThatMightPanic()
        }()
        err := <-done
        if err != nil {
        
            panic(err)
        }
    }

func main() {
        
        var wg conc.WaitGroup
        wg.Go(doSomethingThatMightPanic)
        // panics with a nice stacktrace
        wg.Wait()
    }

### 目标 3：使并发代码更易于阅读 ###

正确地进行并发是很困难的。以一种不混淆代码实际执行的方式进行操作更加困难。该`conc`包试图通过抽象尽可能多的样板复杂性来简化常见操作。

想要使用一组有界的 goroutine 运行一组并发任务？使用 `pool.New()`。想要并发处理有序的结果流，但仍保持顺序？尝试`stream.New()`。切片上的并发映射怎么样？看一看`iter.Map()`。

浏览下面的一些示例，与手动执行这些操作进行一些比较。

## 例子 ##

为了简单起见，这些示例中的每一个都放弃了传播恐慌。要查看会增加什么样的复杂性，请查看上面的“目标 \#2”标题。

生成一组 goroutines 并等待它们完成：

<table> 
 <tbody> 
  <tr> 
   <th><code>stdlib</code></th> 
   <th><code>conc</code></th> 
  </tr> 
  <tr> 
   <td> </td> 
  </tr> 
 </tbody> 
</table>

func main() {
        
        var wg sync.WaitGroup
        for i := 0; i < 10; i++ {
        
            wg.Add(1)
            go func() {
        
                defer wg.Done()
                // crashes on panic!
                doSomething()
            }()
        }
        wg.Wait()
    }

func main() {
        
        var wg conc.WaitGroup
        for i := 0; i < 10; i++ {
        
            wg.Go(doSomething)
        }
        wg.Wait()
    }

在 goroutines 的静态池中处理流的每个元素：

<table> 
 <tbody> 
  <tr> 
   <th><code>stdlib</code></th> 
   <th><code>conc</code></th> 
  </tr> 
  <tr> 
   <td> </td> 
  </tr> 
 </tbody> 
</table>

func process(stream chan int) {
        
        var wg sync.WaitGroup
        for i := 0; i < 10; i++ {
        
            wg.Add(1)
            go func() {
        
                defer wg.Done()
                for elem := range stream {
        
                    handle(elem)
                }
            }()
        }
        wg.Wait()
    }

func process(stream chan int) {
        
        p := pool.New().WithMaxGoroutines(10)
        for elem := range stream {
        
            elem := elem
            p.Go(func() {
        
                handle(elem)
            })
        }
        p.Wait()
    }

在 goroutines 的静态池中处理切片的每个元素：

<table> 
 <tbody> 
  <tr> 
   <th><code>stdlib</code></th> 
   <th><code>conc</code></th> 
  </tr> 
  <tr> 
   <td> </td> 
  </tr> 
 </tbody> 
</table>

func process(values []int) {
        
        feeder := make(chan int, 8)
        var wg sync.WaitGroup
        for i := 0; i < 10; i++ {
        
            wg.Add(1)
            go func() {
        
                defer wg.Done()
                for elem := range feeder {
        
                    handle(elem)
                }
            }()
        }
        for _, value := range values {
        
            feeder <- value
        }
        close(feeder)
        wg.Wait()
    }

func process(values []int) {
        
        iter.ForEach(values, handle)
    }

同时映射一个切片：

<table> 
 <tbody> 
  <tr> 
   <th><code>stdlib</code></th> 
   <th><code>conc</code></th> 
  </tr> 
  <tr> 
   <td> </td> 
  </tr> 
 </tbody> 
</table>

func concMap(
        input []int,
        f func(int) int,
    ) []int {
        
        res := make([]int, len(input))
        var idx atomic.Int64
        var wg sync.WaitGroup
        for i := 0; i < 10; i++ {
        
            wg.Add(1)
            go func() {
        
                defer wg.Done()
                for {
        
                    i := int(idx.Add(1) - 1)
                    if i >= len(input) {
        
                        return
                    }
                    res[i] = f(input[i])
                }
            }()
        }
        wg.Wait()
        return res
    }

func concMap(
        input []int,
        f func(int) int,
    ) []int {
        
        return iter.Map(input, f)
    }

同时处理有序流：

<table> 
 <tbody> 
  <tr> 
   <th><code>stdlib</code></th> 
   <th><code>conc</code></th> 
  </tr> 
  <tr> 
   <td> </td> 
  </tr> 
 </tbody> 
</table>

func mapStream
        in chan int,
        out chan int,
        f func(int) int,
    ) {
        
        tasks := make(chan func())
        taskResults := make(chan chan int)
        // Worker goroutines
        var workerWg sync.WaitGroup
        for i := 0; i < 10; i++ {
        
            workerWg.Add(1)
            go func() {
        
                defer workerWg.Done()
                for task := range tasks {
        
                    task()
                }
            }()
        }
        // Ordered reader goroutines
        var readerWg sync.WaitGroup
        readerWg.Add(1)
        go func() {
        
            defer readerWg.Done()
            for result := range taskResults {
        
                out <- result
            }
        }
        // Feed the workers with tasks
        for elem := range in {
        
            resultCh := make(chan int, 1)
            taskResults <- resultCh
            tasks <- func() {
        
                resultCh <- f(elem)
            }
        }
        // We've exhausted input.
        // Wait for everything to finish
        close(tasks)
        workerWg.Wait()
        close(taskResults)
        readerWg.Wait()
    }

func mapStream(
        in chan int,
        out chan int,
        f func(int) int,
    ) {
        
        s := stream.New().WithMaxGoroutines(10)
        for elem := range in {
        
            elem := elem
            s.Go(func() stream.Callback {
        
                res := f(elem)
                return func() {
         out <- res }
            })
        }
        s.Wait()
    }

## 地位 ##

这个包目前是 pre-1.0。在我们稳定 API 并调整默认值时，在 1.0 版本发布之前可能会有一些小的破坏性变化。如果您有任何问题、疑虑或请求希望在 1.0 版本之前得到解决，请打开一个问题。目前，1.0 的目标是 2023 年 3 月。

[conc_ better structured concurrency for go]: https://github.com/sourcegraph/conc/blob/main/README.md
[03331c245fc34fae864b2640d608162a.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/28/15545055b2444b8584c03a8a12de7ac4.png
[conc.WaitGroup]: https://pkg.go.dev/github.com/sourcegraph/conc#WaitGroup
[pool.Pool]: https://pkg.go.dev/github.com/sourcegraph/conc/pool#Pool
[pool.ResultPool]: https://pkg.go.dev/github.com/sourcegraph/conc/pool#ResultPool
[pool._Result_ErrorPool]: https://pkg.go.dev/github.com/sourcegraph/conc/pool#ErrorPool
[pool._Result_ContextPool]: https://pkg.go.dev/github.com/sourcegraph/conc/pool#ContextPool
[stream.Stream]: https://pkg.go.dev/github.com/sourcegraph/conc/stream#Stream
[iter.Map]: https://pkg.go.dev/github.com/sourcegraph/conc/iter#Map
[iter.ForEach]: https://pkg.go.dev/github.com/sourcegraph/conc/iter#ForEach
[panics.Catcher]: https://pkg.go.dev/github.com/sourcegraph/conc/panics#Catcher
[pool.New]: https://pkg.go.dev/github.com/sourcegraph/conc/pool#New
[pool.NewWithResults_T]: https://pkg.go.dev/github.com/sourcegraph/conc/pool#NewWithResults
[p.WithMaxGoroutines]: https://pkg.go.dev/github.com/sourcegraph/conc/pool#Pool.MaxGoroutines
[p.WithErrors]: https://pkg.go.dev/github.com/sourcegraph/conc/pool#Pool.WithErrors
[p.WithContext_ctx]: https://pkg.go.dev/github.com/sourcegraph/conc/pool#Pool.WithContext
[p.WithFirstError]: https://pkg.go.dev/github.com/sourcegraph/conc/pool#ErrorPool.WithFirstError
[p.WithCollectErrored]: https://pkg.go.dev/github.com/sourcegraph/conc/pool#ResultContextPool.WithCollectErrored
[Link 1]: https://vorpus.org/blog/notes-on-structured-concurrency-or-go-statement-considered-harmful/