Golang 異步編程方式和技巧

Golang 基於多線程、協程實現，與生俱來適合異步編程，當我們遇到那種需要批量處理且耗時的操作時，傳統的線性執行就顯得喫力，這時就會想到異步並行處理。下面介紹一些異步編程方式和技巧。

作者：zvalhu

一、使用方式

1.1、最簡單的最常用的方式：使用 go 關鍵詞

func main() {
 go func() {
  fmt.Println("hello world1")
 }()
 go func() {
  fmt.Println("hello world2")
 }()
}

或者：

func main() {
 go Announce("hello world1")
 go Announce("hello world2")
}
func Announce(message string) {
 fmt.Println(message)
}

使用匿名函數傳遞參數

data := "Hello, World!"
go func(msg string) {
      // 使用msg進行異步任務邏輯處理
      fmt.Println(msg)
}(data)

這種方式不需要考慮返回值問題，如果要考慮返回值，可以使用下面的方式。

1.2、通過 goroutine 和 channel 來實現

ch := make(chan int, 1) // 創建一個帶緩衝的channel
// ch := make(chan int, 0) // 創建一個無緩衝的channel

go func() {
    // 異步任務邏輯
    ch <- result // 將結果發送到channel
    // 異步任務邏輯
    close(ch) // 關閉channel，表示任務完成
}()
// 在需要的時候從channel接收結果
result := <-ch

1.3、使用 sync.WaitGroup

sync.WaitGroup 用於等待一組協程完成其任務。通過 Add() 方法增加等待的協程數量，Done() 方法標記協程完成，Wait() 方法阻塞直到所有協程完成。

var wg sync.WaitGroup

// 啓動多個協程
for i := 0; i < 5; i++ {
    wg.Add(1)
    go func(index int) {
        defer wg.Done()
        // 異步任務邏輯
    }(i)
}

// 等待所有協程完成
wg.Wait()

1.4、使用 errgroup 實現協程組的錯誤處理

如果想簡單獲取協程返回的錯誤，errgroup 包很適合，errgroup 包是 Go 語言標準庫中的一個實用工具，用於管理一組協程並處理它們的錯誤。可以使用 errgroup.Group 結構來跟蹤和處理協程組的錯誤。

var eg errgroup.Group
for i := 0; i < 5; i++ {
    eg.Go(func() error {
     return errors.New("error")
    })

    eg.Go(func() error {
     return nil
    })
}

if err := eg.Wait(); err != nil {
    // 處理錯誤
}

二、一些使用技巧

2.1、使用 channel 的 range 和 close 操作

range 操作可以在接收通道上迭代值，直到通道關閉。可以使用 close 函數關閉通道，以向接收方指示沒有更多的值。

ch := make(chan int)

go func() {
    for i := 0; i < 5; i++ {
        ch <- i // 發送值到通道
    }
    close(ch) // 關閉通道
}()

// 使用range迭代接收通道的值
for val := range ch {
    // 處理接收到的值
}

2.2、使用 select 語句實現多個異步操作的等待

ch1 := make(chan int)
ch2 := make(chan string)

go func() {
    // 異步任務1邏輯
    ch1 <- result1
}()

go func() {
    // 異步任務2邏輯
    ch2 <- result2
}()

// 在主goroutine中等待多個異步任務完成
select {
case res1 := <-ch1:
    // 處理結果1
case res2 := <-ch2:
    // 處理結果2
}

2.3、使用 select 和 time.After() 實現超時控制

如果需要在異步操作中設置超時，可以使用 select 語句結合 time.After() 函數實現。

ch := make(chan int)

go func() {
    // 異步任務邏輯
    time.Sleep(2 * time.Second)
    ch <- result
}()

// 設置超時時間
select {
case res := <-ch:
    // 處理結果
case <-time.After(3 * time.Second):
    // 超時處理
}

2.4、使用 select 和 time.After() 實現超時控制

如果需要在異步操作中設置超時，可以使用 select 語句結合 time.After() 函數實現。

ch := make(chan int)

go func() {
    // 異步任務邏輯
    time.Sleep(2 * time.Second)
    ch <- result
}()

// 設置超時時間
select {
case res := <-ch:
    // 處理結果
case <-time.After(3 * time.Second):
    // 超時處理
}

2.5、使用 time.Tick() 和 time.After() 進行定時操作

time.Tick() 函數返回一個通道，定期發送時間值，可以用於執行定時操作。time.After() 函數返回一個通道，在指定的時間後發送一個時間值。

tick := time.Tick(1 * time.Second) // 每秒執行一次操作

for {
    select {
    case <-tick:
        // 執行定時操作
    }
}

select {
case <-time.After(5 * time.Second):
    // 在5秒後執行操作
}

2.6、使用 sync.Mutex 或 sync.RWMutex 進行併發安全訪問

當多個協程併發訪問共享數據時，需要確保數據訪問的安全性。sync.Mutex 和 sync.RWMutex 提供了互斥鎖和讀寫鎖，用於在訪問共享資源之前進行鎖定，以避免數據競爭。sync.RWMutex 是一種讀寫鎖，可以在多個協程之間提供對共享資源的併發訪問控制。多個協程可以同時獲取讀鎖，但只有一個協程可以獲取寫鎖。

var mutex sync.Mutex
var data int

// 寫操作，使用互斥鎖保護數據
mutex.Lock()
data = 123
mutex.Unlock()

// 讀操作，使用讀鎖保護數據
//RLock()加讀鎖時，如果存在寫鎖，則無法加讀鎖；當只有讀鎖或者沒有鎖時，可以加讀鎖，讀鎖可以加載多個
mutex.RLock()
value := data
mutex.RUnlock()

var rwMutex sync.RWMutex
var sharedData map[string]string

// 讀操作，使用rwMutex.RLock讀鎖保護數據
func readData(key string) string {
    rwMutex.RLock()
    defer rwMutex.RUnlock()
    return sharedData[key]
}

// 寫操作，使用rwMutex.Lock寫鎖保護數據
func writeData(key, value string) {
    rwMutex.Lock()
    defer rwMutex.Unlock()
    sharedData[key] = value
}

注意：sync.Mutex 的鎖是不可以嵌套使用的 sync.RWMutex 的 RLock() 是可以嵌套使用的 sync.RWMutex 的 mu.Lock() 是不可以嵌套的 sync.RWMutex 的 mu.Lock() 中不可以嵌套 mu.RLock()

2.7、使用 sync.Cond 進行條件變量控制

sync.Cond 是一個條件變量，用於在協程之間進行通信和同步。它可以在指定的條件滿足之前阻塞等待，並在條件滿足時喚醒等待的協程。

var cond = sync.NewCond(&sync.Mutex{})
var ready bool

go func() {
    // 異步任務邏輯
    ready = true

    // 通知等待的協程條件已滿足
    cond.Broadcast()
}()

// 在某個地方等待條件滿足
cond.L.Lock()
for !ready {
    cond.Wait()
}
cond.L.Unlock()

2.8、使用 sync.Pool 管理對象池

sync.Pool 是一個對象池，用於緩存和複用臨時對象，可以提高對象的分配和回收效率。

type MyObject struct {
    // 對象結構
}

var objectPool = sync.Pool{
    New: func() interface{} {
        // 創建新對象
        return &MyObject{}
    },
}

// 從對象池獲取對象
obj := objectPool.Get().(*MyObject)

// 使用對象

// 將對象放回對象池
objectPool.Put(obj)

2.9、使用 sync.Once 實現只執行一次的操作

sync.Once 用於確保某個操作只執行一次，無論有多少個協程嘗試執行它，常用於初始化或加載資源等場景。

var once sync.Once
var resource *Resource

func getResource() *Resource {
    once.Do(func() {
        // 執行初始化資源的操作，僅執行一次
        resource = initResource()
    })
    return resource
}

// 在多個協程中獲取資源
go func() {
    res := getResource()
    // 使用資源
}()

go func() {
    res := getResource()
    // 使用資源
}()

2.10、使用 sync.Once 和 context.Context 實現資源清理

可以結合使用 sync.Once 和 context.Context 來確保在多個協程之間只執行一次資源清理操作，並在取消或超時時進行清理。

var once sync.Once

func cleanup() {
    // 執行資源清理操作
}

func doTask(ctx context.Context) {
    go func() {
        select {
        case <-ctx.Done():
            once.Do(cleanup) // 只執行一次資源清理
        }
    }()

    // 異步任務邏輯
}

2.11、使用 sync.Map 實現併發安全的映射

sync.Map 是 Go 語言標準庫中提供的併發安全的映射類型，可在多個協程之間安全地進行讀寫操作。

var m sync.Map

// 存儲鍵值對
m.Store("key", "value")

// 獲取值
if val, ok := m.Load("key"); ok {
    // 使用值
}

// 刪除鍵
m.Delete("key")

2.12、使用 context.Context 進行協程管理和取消

context.Context 用於在協程之間傳遞上下文信息，並可用於取消或超時控制。可以使用 context.WithCancel() 創建一個可取消的上下文，並使用 context.WithTimeout() 創建一個帶有超時的上下文。

ctx, cancel := context.WithCancel(context.Background())

go func() {
    // 異步任務邏輯
    if someCondition {
        cancel() // 取消任務
    }
}()

// 等待任務完成或取消
select {
case <-ctx.Done():
    // 任務被取消或超時
}

2.13、使用 context.WithDeadline() 和 context.WithTimeout() 設置截止時間

context.WithDeadline() 和 context.WithTimeout() 函數可以用於創建帶有截止時間的上下文，以限制異步任務的執行時間。

func doTask(ctx context.Context) {
    // 異步任務邏輯

    select {
    case <-time.After(5 * time.Second):
        // 超時處理
    case <-ctx.Done():
        // 上下文取消處理
    }
}

func main() {
    ctx := context.Background()
    ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
    defer cancel()

    go doTask(ctx)

    // 繼續其他操作
}

2.14、使用 context.WithValue() 傳遞上下文值

context.WithValue() 函數可用於在上下文中傳遞鍵值對，以在協程之間共享和傳遞上下文相關的值。

type keyContextValue string

func doTask(ctx context.Context) {
    if val := ctx.Value(keyContextValue("key")); val != nil {
        // 使用上下文值
    }
}

func main() {
    ctx := context.WithValue(context.Background(), keyContextValue("key"), "value")
    go doTask(ctx)

    // 繼續其他操作
}

2.15、使用 atomic 包進行原子操作

atomic 包提供了一組函數，用於實現原子操作，以確保在併發環境中對共享變量的讀寫操作是原子的。

var counter int64

func increment() {
    atomic.AddInt64(&counter, 1)
}

func main() {
    var wg sync.WaitGroup

    for i := 0; i < 100; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            increment()
        }()
    }

    wg.Wait()
    fmt.Println("Counter:", counter)
}

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/PuNu65ggHyB6jxRqhbN_VQ