在并发处理中,资源争用是一个常见的问题。为了避免资源争用,需要进行优化。以下是一些可以优化并发处理中的资源争用问题的建议:
避免锁竞争:锁竞争是一种常见的资源争用问题。可以通过减少锁的使用,使用更细粒度的锁,以及避免不必要的锁竞争来减少锁竞争。使用缓存:在一些情况下,可以使用缓存来减少资源争用。例如,在处理一些计算密集型的任务时,可以使用缓存来避免重复计算。使用原子操作:原子操作可以在不使用锁的情况下实现资源的同步访问。Go 语言中提供了一些原子操作,例如 atomic.AddInt32 和 atomic.LoadInt32 等,可以用于实现线程安全的资源访问。使用互斥锁:互斥锁是一种用于避免并发资源争用的机制。在需要对资源进行访问的时候,可以使用互斥锁来保证资源的独占访问。使用读写锁:读写锁是一种特殊的互斥锁,可以允许多个读操作同时进行,但是只允许一个写操作进行。在读操作频繁的场景下,可以使用读写锁来提高并发性能。使用条件变量:条件变量是一种用于在不同线程之间进行协调的机制。可以使用条件变量来避免不必要的资源争用。例如,在一个生产者-消费者模式的程序中,可以使用条件变量来协调生产者和消费者之间的交互,从而避免资源争用。但是如果让你不用锁,条件变量,回调的话,还怎么写并发程序啊,谷歌大佬Sameer给了大家一个思路。"Advanced Go Concurrency Patterns" by Sameer Ajmani: 这篇博客深入研究了 Golang 中的并发模式,并讨论了如何使用它们来构建高性能系统。它仅仅使用了Go语言的goroutine和channel便实现高效异步并发编程,没有用到诸如await,context等包括锁,条件变量,和回调函数,文章包括一些示例和实践建议,帮助读者更好地理解和实践这些概念。下面我们针对他给出的case做一些说明与总结,同时对go语言并发编程的语言特性与技巧进行总结,换句话就是说想提炼出面向场景的go语言高并发编程的八股模式。
(资料图)
select-loop的编程关键要素1.如何处理事件
2.如何处理元素
3.如何关闭退出
代码示例:
核心结构与接口
下面代码给出了核心结构sub,以及它实现了接口subscription的关键代码。
updates属性是一个通道,用于用户对元素进行处理。fetcher是用于获取元素的客户端,它可以是从数据库读取,也可以是从消息队列读取。closing用于关闭退出select-loop主体// sub implements the Subscription interface.type sub struct {fetcher Fetcher // fetches itemsupdates chan Item // sends items to the userclosing 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通道中同步写入errcreturn <-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 Updatesclosing: 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 = 10type fetchResult struct {fetched []Itemnext time.Timeerr error}var fetchDone chan fetchResult // if non-nil, Fetch is runningvar pending []Itemvar next time.Timevar err errorvar seen = make(map[string]bool)for {var fetchDelay time.Durationif now := time.Now(); next.After(now) {fetchDelay = next.Sub(now)}var startFetch <-chan time.Timeif fetchDone == nil && len(pending) < maxPending { //等待队列长度未超过最大设置且fetchDone是空,即元素已经都入队列了 // 设置fetchDelay时间后,startFetch通道有值startFetch = time.After(fetchDelay) }var first Itemvar updates chan Itemif 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.errfetched := result.fetchednext, err = result.next, result.errif 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 <- errclose(s.updates)returncase 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 runningvar 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 loopservice channel, reply channels (chan chan error)nil channels in select cases通过err,next,pending三个变量,就实现了在没有锁,条件变量,回调情况下,编写高效并发go程序的需求。