在没有锁，条件变量，回调情况下，如何编写高效异步并发的Go程序-编程学习网

背景

不用锁，条件变量，回调的话，还怎么写并发程序啊，谷歌大佬Sameer给了大家一个思路。"Advanced Go Concurrency Patterns" by Sameer Ajmani: 这篇博客深入研究了 Golang 中的并发模式，并讨论了如何使用它们来构建高性能系统。它包括一些示例和实践建议，帮助读者更好地理解和实践这些概念。下面我们针对他给出的case做一些说明与总结，同时对go语言编程的模式技巧进行总结，换句话就是说想提炼出面向场景的go语言编程的八股模式。

select-loop的编程关键要素

如何处理事件
如何处理元素
如何关闭退出

代码示例:

核心结构与接口

下面代码给出了核心结构sub,以及它实现了接口subscription的关键代码。

updates属性是一个通道，用于用户对元素进行处理。
fetcher是用于获取元素的客户端，它可以是从数据库读取，也可以是从消息队列读取。
closing用于关闭退出select-loop主体.

// sub implements the Subscription interface.
type sub struct {
  fetcher Fetcher         // fetches items
  updates chan Item       // sends items to the user
  closing chan chan error // for Close
}

func (s *sub) Updates() <-chan Item {
  return s.updates
}

func (s *sub) Close() error {
  errc := make(chan error)
  s.closing <- errc // 向closing通道中同步写入errc
  return <-errc     // 等待主loop返回
}

// Subscribe returns a new Subscription that uses fetcher to fetch Items.
func Subscribe(fetcher Fetcher) Subscription {
  s := &sub{
    fetcher: fetcher,
    updates: make(chan Item),       // for Updates
    closing: make(chan chan error), // for Close
  }
  go s.loop()
  return s
}

sub的核心处理逻辑

// loop periodically fecthes Items, sends them on s.updates, and exits
// when Close is called.  It extends dedupeLoop with logic to run
// Fetch asynchronously.
func (s *sub) loop() {
  const maxPending = 10
  type fetchResult struct {
    fetched []Item
    next    time.Time
    err     error
  }
  var fetchDone chan fetchResult // if non-nil, Fetch is running
  var pending []Item
  var next time.Time
  var err error
  var seen = make(map[string]bool)
  for {
    var fetchDelay time.Duration
    if now := time.Now(); next.After(now) {
      fetchDelay = next.Sub(now)
    }
    var startFetch <-chan time.Time
    if fetchDone == nil && len(pending) < maxPending { 
      //等待队列长度未超过最大设置且fetchDone是空，即元素已经都入队列了
      // 设置fetchDelay时间后，startFetch通道有值
      startFetch = time.After(fetchDelay) 
    }
    var first Item
    var updates chan Item
    if len(pending) > 0 {
      first = pending[0]
      updates = s.updates // updates通道是为了用户进一步消费的
    }
    select {
    case <-startFetch:
      fetchDone = make(chan fetchResult, 1)
      go func() {
        fetched, next, err := s.fetcher.Fetch()
        fetchDone <- fetchResult{fetched, next, err}
      }()
    case result := <-fetchDone:
      fetchDone = nil
      // Use result.fetched, result.next, result.err
      fetched := result.fetched
      next, err = result.next, result.err
      if err != nil {
        next = time.Now().Add(10 * time.Second)
        break
      }
      for _, item := range fetched {
        if id := item.GUID; !seen[id] {
          pending = append(pending, item)
          seen[id] = true
        }
      }
    case errc := <-s.closing:
      errc <- err
      close(s.updates)
      return
    case updates <- first:
      pending = pending[1:]
    }
  }
}

那么上面的代码是如何处理三个关键问题的呢？

首先关于关闭并退出loop

上述代码通过监听sub结构的closing属性，实现退出。

//Close asks loop to exit and waits for a response.
func (s *sub) Close() error {
    errc := make(chan error)
    s.closing <- errc
    return <-errc
}

当调用sub的Close方法时，s.closing会接收一个errc的通道，loop主体向errc中写入error信息并退出，调用sub的Close方法的客户端从errc中也同步收到error信息。这是一个同步关闭的过程。loop主体可以在给客户端发送error信息之前，可以完成一系列的关闭清理工作。

关于事件处理与调度

程序中设置的下一次获取元素的延迟调度的最小单位是10秒，从下面第22行可以看到，如果获取元素很快，没有耗费10秒，那么fetchDelay便有个时间gap,startFetch（第7行）这个时间通道便会通过time.After这个方法，在fetchDelay时间后，收到信号，完成18到25行的获取元素工作。

var pending []Item // appended by fetch; consumed by send
    var next time.Time // initially January 1, year 0
    var err error
    for {
        var fetchDelay time.Duration // initially 0 (no delay)
        if now := time.Now(); next.After(now) {
            fetchDelay = next.Sub(now)
        }
        startFetch := time.After(fetchDelay)

     select {
        case <-startFetch:
            var fetched []Item
            fetched, next, err = s.fetcher.Fetch()
            if err != nil {
                next = time.Now().Add(10 * time.Second)
                break
            }
            pending = append(pending, fetched...)
       
        }
    }

问题：为了防止等待队列过大，所以只有当长度不超过maxPending，并且获取的数据已经入队了的时候，才会设置startFetch,否则就不触发fetch。这块可以结合上面整个代码看看。

var fetchDelay time.Duration
        if now := time.Now(); next.After(now) {
            fetchDelay = next.Sub(now)
        }
        var startFetch <-chan time.Time
        if fetchDone == nil && len(pending) < maxPending {
            startFetch = time.After(fetchDelay) // enable fetch case
        }

问题: Loop blocks on Fetch。

golang有个特性，就是Sends and receives on nil channels block.利用这个特性，当fetchDone是nil或者他里面没有准备好结果的时候，相关的case都会阻塞，那么select也不会选择它。同时为了防止fetch函数阻塞loop主函数，通过启动协程（下面9-12行），再次提升主loop的性能。

type fetchResult struct{ fetched []Item; next time.Time; err error }

var fetchDone chan fetchResult // if non-nil, Fetch is running
var startFetch <-chan time.Time
        if fetchDone == nil && len(pending) < maxPending {
            startFetch = time.After(fetchDelay) // enable fetch case
        }
select {
        case <-startFetch:
            fetchDone = make(chan fetchResult, 1)
            go func() {
                fetched, next, err := s.fetcher.Fetch()
                fetchDone <- fetchResult{fetched, next, err}
            }()
        case result := <-fetchDone:
            fetchDone = nil
            // Use result.fetched, result.next, result.err

总结

上面用到了3个技巧，如下所示:

for-select loop
service channel, reply channels (chan chan error)
nil channels in select cases

通过err,next,pending三个变量，就实现了在没有锁，条件变量，回调情况下，编写高效并发go程序的需求。

参考文献：

https://go.dev/talks/2013/advconc.slide

文章详情

在没有锁，条件变量，回调情况下，如何编写高效异步并发的Go程序

背景

select-loop的编程关键要素

sub的核心处理逻辑

总结

软考中级精品资料免费领

相关文章

猜你喜欢

在没有锁，条件变量，回调情况下，如何编写高效异步并发的Go程序

Go语言异步高并发编程的秘密：无锁，无条件变量，无回调