首页技术生活杂记搜索
首页技术生活杂记

golang语言实践

作者：杨润炜
日期：2021/3/14 16:08
go语言

goroutine生命周期管理

使用goroutine需要注意的点：

  1. 启动goroutine后，需要有机制能控制其如何退出，避免goroutine泄漏造成系统不稳定的风险；
  2. 尽量让调用者来决定是否使用goroutine；

注：强烈推荐看看文末引用里关于goroutine leak的文章

反例

下面无法控制goroutine的例子，只能靠log.Fatal调用底层的os.exit退出，但这会使整个进程退出，有些正在处理的状态会因此中断，产生不可预料的异常；
在startServer内使用goroutine，调用者main函数并不知道其使用了goroutine，实际场景下可能会遗漏管理其非阻塞的特征。

  1. package main
  3. import (
  4.     "log"
  5.     "net/http"
  6. )
  8. func main() {
  9.     // some goroutine...
  10.     startApp()
  11.     startDebug()
  12.     select {}
  13. }
  15. func startApp() {
  16.     go func() {
  17.         if err := http.ListenAndServe(":8080", nil); err != nil {
  18.             log.Fatal(err)
  19.         }
  20.     }()
  21. }
  23. func startDebug() {
  24.     go func() {
  25.         if err := http.ListenAndServe(":8081", nil); err != nil {
  26.             log.Fatal(err)
  27.         }
  28.     }()
  29. }

使用chan管理

  1. package main
  3. import (
  4.     "context"
  5.     "fmt"
  6.     "log"
  7.     "net/http"
  8. )
  10. func startServer(addr string, stop <-chan struct{}) error {
  12.     mux := http.NewServeMux()
  13.     mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
  14.         fmt.Fprintf(w, "GET / OK")
  15.     })
  17.     s := http.Server{
  18.         Addr:    addr,
  19.         Handler: mux,
  20.     }
  22.     go func() {
  23.         <-stop
  24.         log.Printf("addr: %s shuntDown", addr)
  25.         s.Shutdown(context.Background())
  26.     }()
  28.     return s.ListenAndServe()
  29. }
  31. func main() {
  32.     stop := make(chan struct{})
  33.     done := make(chan error, 2)
  35.     go func() {
  36.         done <- startServer(":8080", stop) // app
  37.     }()
  38.     go func() {
  39.         done <- startServer(":8081", stop) // debug
  40.     }()
  42.     for i := 0; i < cap(done); i++ {
  43.         if err := <-done; err != nil {
  44.             close(stop)
  45.         }
  46.     }
  48. }

使用context管理

  1. package main
  3. import (
  4.     "context"
  5.     "log"
  6.     "net/http"
  7.     "time"
  8. )
  10. func startServer(ctx context.Context, addr string) error {
  12.     s := http.Server{
  13.         Addr: addr,
  14.     }
  16.     go func(ctx context.Context) {
  17.         <-ctx.Done()
  18.         log.Printf("addr: %s shuntDown", addr)
  19.         s.Shutdown(ctx)
  20.     }(ctx)
  22.     log.Printf("addr: %s start", addr)
  23.     return s.ListenAndServe()
  24. }
  26. func main() {
  28.     ctx, cancel := context.WithCancel(context.Background())
  29.     go func() {
  30.         startServer(ctx, ":8080") // app
  31.     }()
  32.     go func() {
  33.         startServer(ctx, ":8081") // debug
  34.     }()
  36.     cancel()
  37.     select {
  38.     case <-time.After(2 * time.Second):
  39.         log.Print("done")
  40.     }
  41. }

使用errgroup管理

  1. package main
  3. import (
  4.     "context"
  5.     "errors"
  6.     "log"
  7.     "net/http"
  8.     "time"
  10.     "golang.org/x/sync/errgroup"
  11. )
  13. func main() {
  14.     g, ctx := errgroup.WithContext(context.Background())
  15.     g.Go(func() error {
  16.         return startServer(ctx, "3000")
  17.     })
  18.     g.Go(func() error {
  19.         return startServer(ctx, "4000")
  20.     })
  21.     g.Go(func() error {
  22.         time.Sleep(2 * time.Second)
  23.         return errors.New("it's time to exit")
  24.     })
  25.     err := g.Wait()
  26.     log.Printf("exit, err: %v\n", err)
  27. }
  29. func startServer(ctx context.Context, port string) error {
  30.     log.Printf("startServer: %s\n", port)
  31.     srv := http.Server{Addr: ":" + port}
  32.     go func(ctx context.Context) {
  33.         <-ctx.Done()
  34.         if err := srv.Shutdown(ctx); err != nil {
  35.             log.Printf("HTTP server %s Shutdown error: %v", srv.Addr, err)
  36.         }
  37.         log.Printf("HTTP server %s Shutdown successfully", srv.Addr)
  38.     }(ctx)
  40.     return srv.ListenAndServe()
  41. }

错误处理

  1. error是一个值，而不是异常；
  2. panic意味着程序挂了，谨慎使用；
  3. errors.New返回指针类型，详情看error源码；
  4. 使用errors.Wrapf跟踪错误堆栈；

errors.New返回指针的原因

errors.New返回的是指针，能够避免error字符串相同导致错误被误以为一致。

  1. package main
  3. import (
  4.     "errors"
  5.     "log"
  6. )
  8. type myErrorString struct {
  9.     s string
  10. }
  12. func (e myErrorString) Error() string {
  13.     return e.s
  14. }
  16. // New create my error
  17. func New(s string) error {
  18.     return myErrorString{s} // 这里返回的是字符串，而errors.New返回的是指针
  19. }
  21. func main() {
  23.     var err1 = errors.New("test_err")
  24.     var err2 = errors.New("test_err")
  26.     if err1 == err2 {
  27.         log.Println("err1 == err2")
  28.     }
  30.     var err3 = New("test_err")
  31.     var err4 = New("test_err")
  32.     if err3 == err4 {
  33.         log.Println("err3 == err4")
  34.     }
  35. }

用errors.Wrapf跟踪异常的堆栈信息

使用系统内置errors

  1. package main
  3. import (
  4.     "log"
  6.     "errors"
  7. )
  9. // ErrNotFound some data is not found
  10. var ErrNotFound = errors.New("not found")
  12. func dao(id string) (interface{}, error) {
  13.     return nil, errors.New("not found")
  14. }
  16. func service(id string) (interface{}, error) {
  17.     data, err := dao(id)
  18.     if err != nil {
  19.         return nil, err
  20.     }
  21.     return data, nil
  22. }
  24. func api() {
  25.     id := "test"
  26.     data, err := service(id)
  27.     if err != nil {
  28.         // only print: "err: not found"
  29.         log.Printf("err: %+v\n", err)
  30.         return
  31.     }
  32.     log.Printf("found data: %+v\n", data)
  33. }
  35. func main() {
  36.     api()
  37. }

从errors.Wrapf拿到错误堆栈信息

  1. package main
  3. import (
  4.     "log"
  6.     "github.com/pkg/errors"
  7. )
  9. // ErrNotFound some data is not found
  10. var ErrNotFound = errors.New("not found")
  12. func dao(id string) (interface{}, error) {
  13.     return nil, errors.Wrapf(ErrNotFound, "id: %s is not found", id)
  14. }
  16. func service(id string) (interface{}, error) {
  17.     data, err := dao(id)
  18.     if err != nil {
  19.         return nil, err
  20.     }
  21.     return data, nil
  22. }
  24. func api() {
  25.     id := "test"
  26.     data, err := service(id)
  27.     if errors.Cause(err) == ErrNotFound {
  28.         // 打印出错误的堆栈
  29.         log.Printf("stack trace: \n%+v\n", err)
  31.         // 只打印根错误
  32.         log.Printf("original err: %+v\n", errors.Cause(err))
  34.         return
  35.     }
  36.     log.Printf("found data: %+v\n", data)
  37. }
  39. func main() {
  40.     api()
  41. }

context的应用

  1. 级联取消goroutine；
  2. 数据共享、传递；

级联取消goroutine;

0_1615129527189_4ec10931-66fc-4534-b4c1-06c93dc573ab-image.png
见上文中goroutine生命周期管理的【使用context管理】的代码。

数据共享、传递

context可以存储kv数据，且数据可以通过context.WithValue传递给子函数，如果子函数从当前context的kv找不到，会自动递归父级的context查找，直到找到或父级为nil。
0_1615128228780_8a198b9a-cd0f-4cbb-b8fa-ae2fea26193f-image.png 

  1. package main
  3. import (
  4.     "context"
  5.     "fmt"
  6. )
  8. type favContextKey string
  10. var parentkey = favContextKey("parentKey")
  11. var childKey = favContextKey("childKey")
  13. func parent() {
  14.     ctx := context.WithValue(context.Background(), parentkey, "parentVal")
  15.     fmt.Printf("parentFn: parentVal: %+v\n", ctx.Value(parentkey))
  16.     child(context.WithValue(ctx, childKey, "childVal"))
  17. }
  19. func child(ctx context.Context) {
  20.     fmt.Printf("childFn: parentVal: %+v\n", ctx.Value(parentkey))
  21.     fmt.Printf("childFn: childVal: %+v\n", ctx.Value(childKey))
  22. }
  24. func main() {
  25.     parent()
  26. }

依赖管理

一般是用依赖注入的方式，方便单元测试；但依赖也因此要显式地在代码里不断传递，一般是自行进行依赖管理，如果依赖多了就需要不少这样的代码，wire能够减少依赖管理的代理，使其更加简洁。
详情看官方文档示例，比较清晰：https://github.com/google/wire/blob/main/_tutorial/README.md

内存同步的方法与性能比较

  1. 方法：锁（互斥锁、读写锁、原子操作）, chan实现无（显示）阻塞的内存同步；

  2. 由于atomic使用的是copy-on-write的方法在多任务间共享内存，所以不适用于大量数据的共享；

  3. 性能：原子操作> 读写锁 >？互斥锁 >? chan；
    原子操作：使用的是操作系统提供的CAS，不需要进行进程、线程或goroutine的切换；
    读写锁、互斥锁：需要暂停其它goroutine，只保留相关的goroutine，涉及到goroutine的唤醒、暂停及其上下文切换的消耗；
    chan：使用互斥锁实现，但其核心思想在于避免显示加锁，用通信的方式解决内存共享的多goroutine编程问题，因为锁编程的复杂度可能带来死锁、活锁、资源耗尽等风险；

go写并发的核心思想:
>
Share memory by communicating, don’t communicate by sharing memory.

  1. package concurrency
  3. import (
  4.     "sync"
  5.     "sync/atomic"
  6.     "testing"
  7. )
  9. type Config struct {
  10.     a []int
  11. }
  13. var readerCount = 1
  15. func BenchmarkMutexMultipleReaders(b *testing.B) {
  16.     var lastValue uint64
  17.     var lock sync.RWMutex
  18.     cfg := Config{
  19.         a: []int{0, 0, 0, 0, 0, 0, 0},
  20.     }
  22.     var wg sync.WaitGroup
  23.     for n := 0; n < readerCount; n++ {
  24.         wg.Add(1)
  25.         go func() {
  26.             for n := 0; n < b.N; n++ {
  27.                 lock.RLock()
  28.                 atomic.SwapUint64(&lastValue, uint64(cfg.a[0]))
  29.                 lock.RUnlock()
  30.             }
  31.             wg.Done()
  32.         }()
  33.     }
  34.     wg.Wait()
  35. }
  37. func BenchmarkAtomicMultipleReaders(b *testing.B) {
  38.     var lastValue uint64
  39.     var v atomic.Value
  40.     cfg := Config{
  41.         a: []int{0, 0, 0, 0, 0, 0, 0},
  42.     }
  43.     v.Store(cfg)
  45.     var wg sync.WaitGroup
  46.     for n := 0; n < readerCount; n++ {
  47.         wg.Add(1)
  48.         go func() {
  49.             for n := 0; n < b.N; n++ {
  50.                 d := v.Load().(Config)
  51.                 atomic.SwapUint64(&lastValue, uint64(d.a[0]))
  52.             }
  53.             wg.Done()
  54.         }()
  55.     }
  56.     wg.Wait()
  57. }
  59. func BenchmarkMutexOneWriterMultipleReaders(b *testing.B) {
  60.     var lastValue uint64
  61.     var lock sync.RWMutex
  62.     var cfg = Config{
  63.         a: []int{0, 0, 0, 0, 0, 0, 0},
  64.     }
  66.     go func() {
  67.         var i = 0
  68.         for n := 0; n < b.N; n++ {
  69.             i++
  70.             lock.Lock()
  71.             cfg = Config{
  72.                 a: []int{i, i + 1, 0, 0, 0, 0, 0},
  73.             }
  74.             lock.Unlock()
  75.         }
  76.     }()
  78.     var wg sync.WaitGroup
  79.     for n := 0; n < readerCount; n++ {
  80.         wg.Add(1)
  81.         go func() {
  82.             for n := 0; n < b.N; n++ {
  83.                 lock.RLock()
  84.                 atomic.SwapUint64(&lastValue, uint64(cfg.a[0]))
  85.                 lock.RUnlock()
  86.             }
  87.             wg.Done()
  88.         }()
  89.     }
  90.     wg.Wait()
  91. }
  93. func BenchmarkAtomicOneWriterMultipleReaders(b *testing.B) {
  94.     var lastValue uint64
  95.     var v atomic.Value
  96.     var cfg = Config{
  97.         a: []int{0, 0, 0, 0, 0, 0, 0},
  98.     }
  99.     v.Store(cfg)
  101.     go func() {
  102.         var i = 0
  103.         for n := 0; n < b.N; n++ {
  104.             i++
  105.             cfg = Config{
  106.                 a: []int{i, i + 1, 0, 0, 0, 0, 0},
  107.             }
  108.             v.Store(cfg)
  109.         }
  110.     }()
  112.     var wg sync.WaitGroup
  113.     for n := 0; n < readerCount; n++ {
  114.         wg.Add(1)
  115.         go func() {
  116.             for n := 0; n < b.N; n++ {
  117.                 d := v.Load().(Config)
  118.                 atomic.SwapUint64(&lastValue, uint64(d.a[0]))
  119.             }
  120.             wg.Done()
  121.         }()
  122.     }
  123.     wg.Wait()
  124. }
  126. func BenchmarkChanOneWriterMultipleReaders(b *testing.B) {
  127.     rc := make(chan int, 1000)
  128.     b.ResetTimer()
  129.     var lastValue uint64
  130.     var cfg = Config{
  131.         a: []int{0, 0, 0, 0, 0, 0, 0},
  132.     }
  134.     go func() {
  135.         var i = 0
  136.         for msg := range rc {
  137.             i++
  138.             switch msg {
  139.             case 0:
  140.                 cfg = Config{
  141.                     a: []int{i, i + 1, 0, 0, 0, 0, 0},
  142.                 }
  143.             case 1:
  144.                 atomic.SwapUint64(&lastValue, uint64(cfg.a[0]))
  145.             }
  147.         }
  148.     }()
  150.     go func() {
  151.         for n := 0; n < b.N; n++ {
  152.             rc <- 0
  153.         }
  154.     }()
  156.     var wg sync.WaitGroup
  157.     for n := 0; n < readerCount; n++ {
  158.         wg.Add(1)
  159.         go func() {
  160.             for n := 0; n < b.N; n++ {
  161.                 rc <- 1
  162.             }
  163.             wg.Done()
  164.         }()
  165.     }
  166.     wg.Wait()
  167. }

readerCounts为4的结果：

readerCounts为1的结果：

Reference

goroutine-leak
atomic vs mutex
go-trace

感谢您的阅读！
如果看完后有任何疑问，欢迎拍砖。
欢迎转载，转载请注明出处：http://www.yangrunwei.com/a/116.html
邮箱：glowrypauky@gmail.com
QQ: 892413924

上一篇：迟到的2020总结

Copyright© 2015-2016 行之足 Designed byyangrunwei

粤ICP备15106353号