golang之sync包
介绍sync包中常用的方法,
- sync：提供基本的同步原语（比如Mutex、RWMutex、Locker）和 工具类（Once、WaitGroup、Cond、Pool、Map）
- sync/atomic：提供变量的原子操作（基于硬件指令 compare-and-swap）
[Once]
sync.Once 被用于控制变量的初始化，这个变量的读写通常遵循单例模式，满足这三个条件：
- 当且仅当第一次读某个变量时，进行初始化（写操作）
- 变量被初始化过程中，所有读都被阻塞（读操作；当变量初始化完成后，读操作继续进行
- 变量仅初始化一次，初始化完成后驻留在内存里
once.Sync 可用于任何符合 "exactly once" 语义的场景，比如：
- 初始化 rpc/http client
- open/close 文件
- close channel
- 线程池初始化
[WaitGroup]
一个 WaitGroup 对象可以等待一组协程结束。使用方法是：
- main协程通过调用
wg.Add(delta int)设置worker协程的个数，然后创建worker协程； - worker协程执行结束以后，都要调用
wg.Done()； - main协程调用
wg.Wait()且被block，直到所有worker协程全部执行结束后返回。
实例:
func main() { // 省略部分代码 ... var wg sync.WaitGroup for _, task := range tasks { task := task wg.Add(1) go func() { task() defer wg.Done() }() } wg.Wait() // 省略部分代码... }
说明:
1.wg.Done 必须在wg.Add之后执行
2.wg.Done在worker协程中调用, 保证调用一次,不能因为panic或者其他原因导致没有执行(建议使用defer wg.Done)
3.task := task 需要进行再次赋值,否则会读取到最后一个元素的值(
- for-loop 内创建的局部变量，即便名字相同，内存地址也不会复用
)
适用场景:
1.需要向多个服务请求数据,并最终将这些服务请求结果进行最终整合返回给前端进行展示
基于WaitGroup的使用,发现一个协程出现错误的时候,并不会将子协程的错误抛送给主goroutine : golang.org/x/sync/errgroup
使用示例:
package main import ( "fmt" "net/http" "golang.org/x/sync/errgroup" ) func main() { var urls = []string{ "http://www.golang.org/", "http://www.baidu.com/", "http://www.bokeyuan12111.com/", } g := new(errgroup.Group) for _, url := range urls { url := url g.Go(func() error { resp, err := http.Get(url) if err != nil { fmt.Println(err) return err } fmt.Printf("get [%s] success: [%d] \n", url, resp.StatusCode) return resp.Body.Close() }) } if err := g.Wait(); err != nil { fmt.Println(err) } else { fmt.Println("All success!") } }
输出:
get [http://www.baidu.com/] success: [200]
Get "http://www.bokeyuan12111.com/": dial tcp: lookup www.bokeyuan12111.com: no such host
Get "http://www.golang.org/": dial tcp 142.251.42.241:80: i/o timeout
Get "http://www.bokeyuan12111.com/": dial tcp: lookup www.bokeyuan12111.com: no such host
可以看到，执行获取www.bokeyuan12111.com和www.golang.org两个 url 的子 groutine 均发生了错误，在主任务 goroutine 中成功捕获到了第一个错误信息。
除了 拥有 WaitGroup 的控制能力 和 错误传播 的功能之外，errgroup 还有最重要的 context 反向传播机制
带有上下文取消的使用方式:
package main import ( "context" "fmt" "golang.org/x/sync/errgroup" ) func main() { g, ctx := errgroup.WithContext(context.Background()) dataChan := make(chan int, 20) // 数据生产端任务子 goroutine g.Go(func() error { defer close(dataChan) for i := 1; ; i++ { if i == 10 { return fmt.Errorf("data 10 is wrong") } dataChan <- i fmt.Println(fmt.Sprintf("sending %d", i)) } }) // 数据消费端任务子 goroutine for i := 0; i < 3; i++ { g.Go(func() error { for j := 1; ; j++ { select { case <-ctx.Done(): return ctx.Err() case number := <-dataChan: fmt.Println(fmt.Sprintf("receiving %d", number)) } } }) } // 主任务 goroutine 等待 pipeline 结束数据流 err := g.Wait() if err != nil { fmt.Println(err) } fmt.Println("main goroutine done!") }
在以上示例中，我们模拟了一个数据传送管道。在数据的生产与消费任务集中，有四个子任务 goroutine：一个生产数据的 goroutine，三个消费数据的 goroutine。当数据生产方存在错误数据时（数据等于 10 ），我们停止数据的生产与消费，并将错误抛出，回到 main goroutine 的执行逻辑中。
可以看到，因为 errgroup 中的 Context cancle 函数的嵌入，我们在子任务 goroutine 中也能反向控制任务上下文。
程序的某一次运行，输出结果如下：
sending 1 sending 2 sending 3 sending 4 sending 5 sending 6 sending 7 sending 8 sending 9 receiving 1 receiving 3 receiving 2 receiving 4 data 10 is wrong main goroutine done!
[Cond]
[Pool]
[Map]
[sync/atomic] 原子相关操作