sourcegraph/conc
GitHub: sourcegraph/conc
conc 是一个 Go 语言结构化并发库，通过提供更安全的 WaitGroup、线程池和并发迭代等抽象，让并发代码更简洁、更不容易泄漏 goroutine。
Stars: 10417 | Forks: 359
# `conc`：更好的 Go 结构化并发
[](https://pkg.go.dev/github.com/sourcegraph/conc)
[](https://sourcegraph.com/github.com/sourcegraph/conc)
[](https://goreportcard.com/report/github.com/sourcegraph/conc)
[](https://codecov.io/gh/sourcegraph/conc)
[](https://discord.gg/bvXQXmtRjN)
`conc` 是你的 Go 结构化并发工具带，让常见任务
更简单、更安全。
```
go get github.com/sourcegraph/conc
```
# 概览
- 如果你只想要一个更安全的 `sync.WaitGroup` 版本，请使用 [`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)
- 如果你想并发地对切片进行 map 操作，请使用 [`iter.Map`](https://pkg.go.dev/github.com/sourcegraph/conc/iter#Map)
- 如果你想并发地遍历切片，请使用 [`iter.ForEach`](https://pkg.go.dev/github.com/sourcegraph/conc/iter#ForEach)
- 如果你想在自己的 goroutine 中捕获 panic，请使用 [`panics.Catcher`](https://pkg.go.dev/github.com/sourcegraph/conc/panics#Catcher)
所有的 pool 都是通过
[`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) 配置池中 goroutine 的最大数量
- [`p.WithErrors()`](https://pkg.go.dev/github.com/sourcegraph/conc/pool#Pool.WithErrors) 配置池以运行返回 error 的任务
- [`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) 配置 error 池，使其仅保留首个返回的 error 而不是聚合的 error
- [`p.WithCollectErrored()`](https://pkg.go.dev/github.com/sourcegraph/conc/pool#ResultContextPool.WithCollectErrored) 配置 result 池，使其在任务出错时也能收集结果
# 目标
该包的主要目标是：
1. 让 goroutine 泄漏变得更加困难
2. 优雅地处理 panic
3. 让并发代码更易读
## 目标 #1：让 goroutine 泄漏变得更加困难
使用 goroutine 时的一个常见痛点是清理它们。很容易启动一个
`go` 语句却未能正确等待它完成。
`conc` 采取了强制性的立场，认为所有并发都应该是有作用域的。
也就是说，goroutine 应该有一个所有者，并且该所有者应该始终
确保其拥有的 goroutine 正确退出。
在 `conc` 中，goroutine 的所有者始终是 `conc.WaitGroup`。goroutine
通过 `(*WaitGroup).Go()` 在 `WaitGroup` 中启动，并且
在 `WaitGroup` 离开作用域之前，应始终调用 `(*WaitGroup).Wait()`。
在某些情况下，你可能希望启动的 goroutine 的生命周期长于调用者的
作用域。在这种情况下，你可以将 `WaitGroup` 传递给启动函数。
```
func main() {
var wg conc.WaitGroup
defer wg.Wait()
startTheThing(&wg)
}
func startTheThing(wg *conc.WaitGroup) {
wg.Go(func() { ... })
}
```
有关为什么作用域并发很好的更多讨论，请查看
[这篇
博客
文章](https://vorpus.org/blog/notes-on-structured-concurrency-or-go-statement-considered-harmful/)。
## 目标 #2：优雅地处理 panic
在长时间运行的应用程序中，goroutine 的一个常见问题是处理
panic。在没有 panic 处理程序的情况下启动的 goroutine 在发生 panic 时会导致整个进程崩溃。
这通常是不希望发生的。
然而，如果你确实为 goroutine 添加了 panic 处理程序，一旦你捕获了
panic，你会怎么处理它？一些选项：
1. 忽略它
2. 记录它
3. 将其转换为 error 并返回给 goroutine 启动器
4. 将 panic 传播给 goroutine 启动器
忽略 panic 是个坏主意，因为 panic 通常意味着确实
出了问题，需要有人来修复它。
仅仅记录 panic 也不是很好，因为这样启动器就不会收到
发生不好事情的指示，即使你的
程序处于非常糟糕的状态，它也可能会像正常情况一样继续运行。
(3) 和 (4) 都是合理的选项，但两者都要求 goroutine 有一个
能够真正接收到出问题消息的所有者。这在使用 `go` 启动的 goroutine 中通常是不成立的，但在 `conc`
包中，所有 goroutine 都有一个必须收集该已启动 goroutine 的所有者。
在 conc 包中，如果任何启动的 goroutine 发生了 panic，任何对 `Wait()` 的调用都会引发 panic。此外，它还会使用来自子
goroutine 的 stacktrace 装饰 panic 值，这样你就不会丢失关于导致 panic 原因的信息。
每次使用 `go` 启动内容时都正确地完成这一切并非
易事，并且它需要大量的样板代码，这使得代码中重要的部分
更难阅读，因此 `conc` 替你完成了这项工作。
stdlib |
conc |
|
```
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()`。那么对
切片进行并发 map 呢？看一下 `iter.Map()`。
浏览下面的一些示例，看看它们与手动完成这些操作的对比。
# 示例
为了简单起见，这些示例都省略了 panic 的传播。要查看
这会增加什么样的复杂性，请查看上面的“目标 #2”标题。
启动一组 goroutine 并等待它们完成：
stdlib |
conc |
|
```
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()
}
```
|
在静态的 goroutine 池中处理流的每个元素：
stdlib |
conc |
|
```
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()
}
```
|
在静态的 goroutine 池中处理切片的每个元素：
stdlib |
conc |
|
```
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)
}
```
|
并发地对切片进行 map：
stdlib |
conc |
|
```
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)
}
```
|
并发地处理有序流：
stdlib |
conc |
|
```
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 {
item := <-result
out <- item
}
}()
// 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()
}
```
|
# 状态
该包目前处于 1.0 版本之前。在稳定 API 和调整默认值之后、1.0 版本发布之前，可能会有一些微小的破坏性
变更。
如果你有任何想在 1.0 版本发布前解决的问题、疑虑或请求，请提出 issue。目前，1.0 版本的目标定于
2023 年 3 月。
标签：EVTX分析, Go, Ruby工具, 并发编程, 开发库, 日志审计, 结构化并发