当前位置：首页 > news >正文

更友好的并发库conc介绍

news 2025/8/14 16:15:03

要在Go语言中实现并发太容易了，对于初学者来说也容易掉入并发陷阱。社区中经验老道的Gopher为我们封装了一个并发工具包——conc，使用它可以轻松应对绝大多数并发场景。

conc 是由sourcegraph开源的一套友好的结构化并发工具包，其中总结了 sourcegraph 内部在编写并发代码时反复遇到的问题的解决方案。

conc.WaitGroup

conc库中的WaitGroup是作用域并发的主要构建块。调用其Go方法能够生成goroutine，调用其Wait方法可以确保在生成的goroutine都退出后再继续执行。这一些特性和标准库中的sync.WaitGroup一样。区别的地方在于conc.WaitGroup中子goroutine中的panic会被传递给Wait方法的调用方，这也正是我们需要conc.WaitGroup的原因，要不然我们就需要自己去recover子goroutine中的panic。

conc.WaitGroup的使用非常简单，可以参考以下代码示例。

// waitGroupDemo conc.WaitGroup并发示例
func waitGroupDemo() {var count atomic.Int64var wg conc.WaitGroup// 开启10个goroutine并发执行 count.Add(1)for i := 0; i < 10; i++ {wg.Go(func() {count.Add(1)})}// 等待10个goroutine都执行完wg.Wait()fmt.Println(count.Load())
}

如果想要自动recover子goroutine可能传递出的panic，可以使用其WaitAndRecover方法。示例代码如下。

// waitGroupDemo 自动recover示例
func waitGroupDemo2() {var count atomic.Int64var wg conc.WaitGroup// 开启10个goroutine并发执行 count.Add(1)for i := 0; i < 10; i++ {wg.Go(func() {if i == 7 {panic("bad thing")}count.Add(1)})}// 等待10个goroutine都执行完wg.WaitAndRecover()fmt.Println(count.Load())
}

扩展：为什么要使用作用域并发？

有一种观点认为：Go语言通过go关键字随时随地创建goroutine发起并发就像在代码中使用goto一样，让程序显得混乱和令人困惑。在并发代码中无论是出现panic还是出现其他错误，为了避免引发更大问题，通常的解决方式是将异常或者堆栈信息传递给调用方，这就需要有堆栈信息和明确的"调用者"为前提，所以我们应该使用结构化的并发编程——把goroutine放到托儿所里，而不是让他们像野孩子一样到处乱跑。

推荐阅读👉Notes on structured concurrency, or: Go statement considered harmful

goroutine池

Pool 是一个用于并发执行任务的 goroutines 池。conc包中针对不同的业务场景定义了以下几种goroutine池。

pool.Pool

使用New()创建一个池对象，然后通过调用Go()提交要执行的任务。提交完所有任务后，必须调用Wait()来清理任何派生的goroutines并传播可能出现的panic。

Pool中的goroutine是延迟启动的（用到的时候再启动），所以创建一个新的Pool是廉价的。产生的 goroutine 永远不会比提交的任务多。

创建Pool是高效的，但也不是零成本。它不适用于耗时非常短的任务。启动和拆卸的开销约为1µs，每个任务的开销约300ns。

对于创建得到的池对象，可以使用With系列函数进行配置。其中最常用的便是通过WithMaxGoroutines()指定池中最大goroutine数量。

例如下面的示例中使用WithMaxGoroutines(3)配置最大goroutine数量为3。

// poolDemo goroutine池示例
func poolDemo() {// 创建一个最大数量为3的goroutine池p := pool.New().WithMaxGoroutines(3)// 使用p.Go()提交5个任务for i := 0; i < 5; i++ {p.Go(func() {fmt.Println("Q1mi")})}p.Wait()
}

注意：对Pool使用Go()提交任务后不允许再调用With系列方法进行配置。

pool.ContextPool

使用WithContext可以创建一个传递Context的Pool，通过这个父Context来控制池中的goroutine。默认情况下，在取消父Context之前，Pool中的Context不会取消。

想要在任何任务返回错误或出现panic时立即取消其Context，可以通过WithCancelOnError进行配置。

// poolWithContextDemoCancelOnError 支持context的池
// goroutine中出错时取消context
func poolWithContextDemoCancelOnError() {p := pool.New().WithMaxGoroutines(4).WithContext(context.Background()).WithCancelOnError() // 出错时取消所有goroutine// 提交3个任务for i := 0; i < 3; i++ {i := ip.Go(func(ctx context.Context) error {if i == 2 {return errors.New("cancel all other tasks")}<-ctx.Done()return nil})}err := p.Wait()fmt.Println(err)
}

pool.WithErrors

当提交的任务有可能返回错误时，可以使用WithErrors得到一个ErrorPool，并通过Wait()获取可能返回的错误。

func poolWithError() {p := pool.New().WithErrors()for i := 0; i < 3; i++ {i := ip.Go(func() error {if i == 2 {return errors.New("oh no!")}return nil})}err := p.Wait()fmt.Println(err)
}

pool.ResultPool

ResultPool是一个执行返回泛型结果的任务池。使用Go()在池中执行任务，然后由Wait()返回任务的结果。

// poolWithResult 执行返回结果的任务池
func poolWithResult() {// 创建一个任务池，其中任务返回的结果为intp := pool.NewWithResults[int]()for i := 0; i < 10; i++ {i := ip.Go(func() int {return i * 2})}res := p.Wait()// 结果的顺序是不确定的, 所以这里先排序再打印sort.Ints(res)fmt.Println(res)
}

pool.ResultContextPool

ResultContextPool中执行的任务接受一个context参数并返回结果。

pool.ResultErrorPool

ResultErrorPool中执行的任务会返回一个泛型结果和错误。

Stream

在Pool中并发执行任务后返回结果是无序的，conc中有一个stream包提供任务并发、结果有序的实现，适用于在维护结果顺序的同时并发执行任务流。

使用Stream时提交的每个任务都返回一个回调函数。每个任务都将在任务池中同时执行，但是回调函数将按照任务提交的顺序依次执行。

等到所有任务都提交后，必须调用 Wait()来等待正在运行的 goroutine 运行完。当任务执行过程中或回调函数执行期间出现 panic 时，所有其他任务和回调仍将执行。当调用 Wait()时，panic将传给调用方。

同Pool一样，Stream也不适用于非常短的任务。启动和拆卸会增加几微秒的开销，每个任务的开销大约是500ns。对于任何需要网络通话的任务来说，这性能都足够好了。

// streamDemo 并发的流式任务示例
func streamDemo() {times := []int{20, 52, 16, 45, 4, 80}s := stream.New()for _, millis := range times {dur := time.Duration(millis) * time.Millisecond// 提交任务s.Go(func() stream.Callback {time.Sleep(dur)// 虽然上一行通过sleep增加了时间// 但最终结果仍按任务提交（s.Go）的顺序打印return func() { fmt.Println(dur) }})}s.Wait()
}

输出结果：

20ms
52ms
16ms
45ms
4ms
80ms

iter

conc包还提供了一个iter。在iter包中提供了基于泛型实现的iterator和mapper，封装了以下四种日常开发中常用的迭代方法。

func ForEach[T any](input []T, f func(*T))
func ForEachIdx[T any](input []T, f func(int, *T))
func Map[T, R any](input []T, f func(*T) R) []R
func MapErr[T, R any](input []T, f func(*T) (R, error)) ([]R, error)

初始状态下可直接调用iter包的上述函数。

// iterDemo 迭代器ForEach示例
func iterDemo() {input := []int{1, 2, 3, 4}// 可直接调用iter包的ForEach函数iter.ForEach(input, func(v *int) {if *v%2 != 0 {*v = -1}})fmt.Println(input)
}

如果想要更定制化的功能，可以自行构造iterator或mapper。

iterator示例

iterator适用于对序列中的每个元素执行统一操作的场景。

// iteratorDemo 迭代器示例
func iteratorDemo() {input := []int{1, 2, 3, 4}// 创建一个最大goroutine个数为输入元素一半的迭代器iterator := iter.Iterator[int]{MaxGoroutines: len(input) / 2,}iterator.ForEach(input, func(v *int) {if *v%2 != 0 {*v = -1}})fmt.Println(input)
}

输出：

[-1 2 -1 4]

mapper示例

mapper是一个带有结果类型的迭代器，适用于遍历序列中的每个元素执行统一操作后拿到返回结果的场景。

// mapperDemo mapper示例
func mapperDemo() {input := []int{1, 2, 3, 4}// 创建一个最大goroutine个数为输入元素一半的映射器mapper := iter.Mapper[int, bool]{MaxGoroutines: len(input) / 2,}results := mapper.Map(input, func(v *int) bool { return *v%2 == 0 })fmt.Println(results)
}

输出：

[false true false true]

panics

conc下提供了一个处理panic相关的panics包。其中panics.Catcher是用来捕获任务运行时可能出现的panic。具体使用方法是通过 Try() 执行任务函数，该函数将捕获任何产生的panic。Try()可以在任意 goroutine 中被调用任意次数。一旦所有调用完成后，使用 Recovered()获取第一个panic(如果有的话)的值，或者使用 Repanic()传递panic(重新panic)。

// panicDemo recover可能出现的异常
func panicDemo() {var pc panics.Catcheri := 0pc.Try(func() { i += 1 })pc.Try(func() { panic("abort!") })pc.Try(func() { i += 1 })// recover可能出现的panicrc := pc.Recovered()// 重新panic// pc.Repanic()fmt.Println(i)fmt.Println(rc.Value.(string))
}