八. Go并发编程--errGroup
<h2 id="一-前言">一. 前言</h2><p>了解 <code>sync.WaitGroup</code>的用法都知道</p>
<ul>
<li>
<p>一个 <code>goroutine</code> 需要等待多个 <code>goroutine</code> 完成和多个 <code>goroutine</code> 等待一个 <code>goroutine</code> 干活时都可以解决问题</p>
<br>
</li>
</ul>
<p><code>WaitGroup</code> 的确是一个很强大的工具，但是使用它相对来说还是有一点小麻烦，</p>
<ul>
<li>一方面我们需要自己手动调用 <code>Add()</code> 和 Done() 方法，一旦这两个方法有一个多调用或者少调用，最终都有可能导致程序崩溃，所以我们在使用这两个方法的时候要格外小心，确保最终计数器能够达到 0 的状态；</li>
<li>另一方面就是它不能抛出错误给调用者，只要一个 goroutine 出错我们就不再等其他 goroutine 了，减少资源浪费，所以我们只能通过声明多个外部变量的方式（或者声明一个变量然后通过加锁来更新它的值）来分别接收每个协程的 error 才行，就像下面的代码：<pre><code class="language-golang">func main() {
var (
wg sync.WaitGroup
err1, err2 error// 通过在外部定义变量用来记录错误
)
wg.Add(2)
go func() {
defer wg.Done()
fmt.Print("task 1")
err1 = nil
}()
go func() {
defer wg.Done()
fmt.Print("task 2")
err2 = fmt.Errorf("task 2 error")
}()
wg.Wait()
if err1 != nil || err2 != nil {
// TODO
}
fmt.Print("finish")
}
</code></pre>
</li>
</ul>
<p>使用<code>WaitGroup</code> 都不能很好的解决。 所以此时可以使用 <code>ErrGroup</code> 就可以解决问题了。</p>
<br>
<h2 id="二-errgroup">二. Errgroup</h2>
<p><code>Errgroup</code> 是 Golang 官方提供的一个同步扩展库, 代码仓库如下</p>
<ul>
<li>errgroup</li>
</ul>
<p>它和 <code>WaitGroup</code> 的作用类似，但是它提供了更加丰富的功能以及更低的使用成本：</p>
<ol>
<li>和context集成；</li>
<li>能够对外传播error,可以把子任务的错误传递给<code>Wait</code> 的调用者.</li>
</ol>
<br>
<p><code>Errgroup</code> 的代码非常简短，加上注释一共才 66 行，包含一个结构体以及三个对外暴露的方法，接下来就让我们走进源码，来具体看一下它是如何工作的</p>
<br>
<h3 id="21-group">2.1 Group</h3>
<pre><code class="language-golang">type Group struct {
// context 的 cancel 方法
cancel func()
// 复用 WaitGroup
wg sync.WaitGroup
// 用来保证只会接受一次错误
errOnce sync.Once
// 保存第一个返回的错误
err error
}
</code></pre>
<br>
<h3 id="22-withcontext">2.2 WithContext</h3>
<pre><code class="language-golang">func WithContext(ctx context.Context) (*Group, context.Context) {
// 使用 contex.WithCancel创建一个可以取消的 context 将 cancel 赋值给 Group 保存起来
ctx, cancel := context.WithCancel(ctx)
return &Group{cancel: cancel}, ctx
}
</code></pre>
<p><code>WithContext</code> 就是使用 <code>WithCancel</code>创建一个可以取消的 context 将 cancel 赋值给 Group 保存起来，然后再将 context 返回回去</p>
<blockquote>
<p>注意这里有一个坑，在后面的代码中不要把这个 ctx 当做父 context 又传给下游，因为 errgroup 取消了，这个 context 就没用了，会导致下游复用的时候出错<br>
Go</p>
</blockquote>
<br>
<h3 id="23-go">2.3 Go</h3>
<p>Go 方法传入一个 func() error 内部会启动一个 <code>goroutine</code> 去处理</p>
<pre><code class="language-golang">// Go calls the given function in a new goroutine.
//
// The first call to return a non-nil error cancels the group; its error will be
// returned by Wait.
func (g *Group) Go(f func() error) {
// wg.Add(1) 计数器加 1
g.wg.Add(1)
go func() {
defer g.wg.Done()
if err := f(); err != nil {
// 这里使用sync.Once作用，保证传入的 无入参函数执行一次
// 如果有 error，则记录发生的第一个 error
g.errOnce.Do(func() {
g.err = err
if g.cancel != nil {
g.cancel()
}
})
}
}()
}
</code></pre>
<p>Go 方法其实就类似于 go 关键字，会启动一个协程，然后利用 <code>waitgroup</code> 来控制是否结束，如果有一个非 nil 的 error 出现就会保存起来并且如果有 cancel 就会调用 <code>cancel</code> 取消掉，使 ctx 返</p>
<br>
<h3 id="24-wait">2.4 Wait</h3>
<pre><code class="language-golang">/ Wait blocks until all function calls from the Go method have returned, then
// returns the first non-nil error (if any) from them.
func (g *Group) Wait() error {
// wg.Wait() 等待所有任务执行完毕
g.wg.Wait()
if g.cancel != nil {
g.cancel()
}
return g.err
}
</code></pre>
<p>Wait 方法其实就是调用 WaitGroup 等待，如果有 cancel 就调用一下</p>
<br>
<h2 id="三-案例">三. 案例</h2>
<h3 id="31-记录错误">3.1 记录错误</h3>
<p>使用 Errgroup，上述代码可以改为下面的样子：</p>
<pre><code class="language-golang">package main
import (
"fmt"
"golang.org/x/sync/errgroup"
)
func main() {
var eg errgroup.Group
//匿名函数将会通过GO关键字启动一个协程
eg.Go(func() error {
fmt.Print("task 1\n")
return nil
})
eg.Go(func() error {
fmt.Print("task 2\n")
return fmt.Errorf("task 2 error")
})
// 使用Wait 等待所有的协程执行完毕后，再进行后面的逻辑，同时可以记录两个协程的错误
if err := eg.Wait(); err != nil {
fmt.Printf("some error occur: %s\n", err.Error())
}
fmt.Print("over")
}
</code></pre>
<p>代码简洁了很多</p>
<br>
<h3 id="32-一个协程出错其他协程终止">3.2 一个协程出错，其他协程终止</h3>
<p>基于 <code>errgroup</code> 实现一个 <code>http server</code> 的启动和关闭 ，以及 linux signal 信号的注册和处理，要保证能够 一个退出，全部注销退出。</p>
<pre><code class="language-golang">package main
import (
"context"
"fmt"
"log"
"net/http"
"os"
"os/signal"
"syscall"
"time"
"golang.org/x/sync/errgroup"
)
func main() {
g, ctx := errgroup.WithContext(context.Background())
mux := http.NewServeMux()
mux.HandleFunc("/ping", func(w http.ResponseWriter, r *http.Request) {
w.Write([]byte("pong"))
})
// 模拟单个服务错误退出
serverOut := make(chan struct{})
mux.HandleFunc("/shutdown", func(w http.ResponseWriter, r *http.Request) {
serverOut <- struct{}{}
})
server := http.Server{
Handler: mux,
Addr: ":8080",
}
// g1
// g1 退出了所有的协程都能退出么？
// g1 退出后, context 将不再阻塞，g2, g3 都会随之退出
// 然后 main 函数中的 g.Wait() 退出，所有协程都会退出
g.Go(func() error {
return server.ListenAndServe()
})
// g2
// g2 退出了所有的协程都能退出么？
// g2 退出时，调用了 shutdown，g1 会退出
// g2 退出后, context 将不再阻塞，g3 会随之退出
// 然后 main 函数中的 g.Wait() 退出，所有协程都会退出
g.Go(func() error {
select {
case <-ctx.Done():
log.Println("errgroup exit...")
case <-serverOut:
log.Println("server will out...")
}
timeoutCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
// 这里不是必须的，但是如果使用 _ 的话静态扫描工具会报错，加上也无伤大雅
defer cancel()
log.Println("shutting down server...")
return server.Shutdown(timeoutCtx)
})
// g3
// g3 捕获到 os 退出信号将会退出
// g3 退出了所有的协程都能退出么？
// g3 退出后, context 将不再阻塞，g2 会随之退出
// g2 退出时，调用了 shutdown，g1 会退出
// 然后 main 函数中的 g.Wait() 退出，所有协程都会退出
g.Go(func() error {
quit := make(chan os.Signal, 0)
signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
select {
case <-ctx.Done():
return ctx.Err()
case sig := <-quit:
return fmt.Errorf("get os signal: %v", sig)
}
})
fmt.Printf("errgroup exiting: %+v\n", g.Wait())
}
</code></pre>
<p>运行后，使用 <code>ctrl + C</code> 终止程序，终端输出如下</p>
<pre><code class="language-powershell">2021/11/03 10:20:34 errgroup exit...
2021/11/03 10:20:34 shutting down server...
errgroup exiting: get os signal: interrupt
</code></pre>
<p>这里主要用到了 errgroup 一个出错，其余取消的能力</p>
<br>
<h2 id="四-总结">四. 总结</h2>
<p>当然除了 Golang 官方提供的扩展库之外，还有很多类似的其他优秀开源工具，例如 bilibili/errgroup，支持设置固定数量的协程数以及失败 cancel 机制和 panic-recover 机制等等，感兴趣的同学可以自行去了解一番。</p>
<br>
<h2 id="五-参考">五. 参考</h2>
<ol>
<li>
<p>https://lailin.xyz/post/go-training-week3-errgroup.html</p>
</li>
<li>
<p>https://juejin.cn/post/6996300205989560333</p>
</li>
<li>
<p>https://github.com/golang/sync/tree/master/errgroup</p>
</li>
</ol>
</div>
<div id="MySignature" role="contentinfo">
♥永远年轻，永远热泪盈眶♥<br><br>
来源：https://www.cnblogs.com/failymao/p/15522374.html
頁:
[1]