Go語言異步高并發編程的秘密：無鎖，無條件變量，無回調
背景
在并發處理中，資源爭用是一個常見的問題。為了避免資源爭用，需要進行優化。以下是一些可以優化并發處理中的資源爭用問題的建議：
- 避免鎖競爭：鎖競爭是一種常見的資源爭用問題。可以通過減少鎖的使用，使用更細粒度的鎖，以及避免不必要的鎖競爭來減少鎖競爭。
- 使用緩存：在一些情況下，可以使用緩存來減少資源爭用。例如，在處理一些計算密集型的任務時，可以使用緩存來避免重復計算。
- 使用原子操作：原子操作可以在不使用鎖的情況下實現資源的同步訪問。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 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程序的需求。
