Golang異步編程方式和技巧
作者 | zvalhu
Golang基于多線程、協(xié)程實(shí)現(xiàn)，與生俱來適合異步編程，當(dāng)我們遇到那種需要批量處理且耗時(shí)的操作時(shí)，傳統(tǒng)的線性執(zhí)行就顯得吃力，這時(shí)就會想到異步并行處理。下面介紹一些異步編程方式和技巧。
一、使用方式
1.最簡單的最常用的方式：使用go關(guān)鍵詞
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)
}使用匿名函數(shù)傳遞參數(shù)
data := "Hello, World!"
go func(msg string) {
// 使用msg進(jìn)行異步任務(wù)邏輯處理
fmt.Println(msg)
}(data)這種方式不需要考慮返回值問題，如果要考慮返回值，可以使用下面的方式。
2.通過goroutine和channel來實(shí)現(xiàn)
ch := make(chan int, 1) // 創(chuàng)建一個帶緩沖的channel
// ch := make(chan int, 0) // 創(chuàng)建一個無緩沖的channel
go func() {
// 異步任務(wù)邏輯
ch <- result // 將結(jié)果發(fā)送到channel
// 異步任務(wù)邏輯
close(ch) // 關(guān)閉channel，表示任務(wù)完成
}()
// 在需要的時(shí)候從channel接收結(jié)果
result := <-ch3.使用sync.WaitGroup
sync.WaitGroup用于等待一組協(xié)程完成其任務(wù)。通過Add()方法增加等待的協(xié)程數(shù)量，Done()方法標(biāo)記協(xié)程完成，Wait()方法阻塞直到所有協(xié)程完成。
var wg sync.WaitGroup
// 啟動多個協(xié)程
for i := 0; i < 5; i++ {
wg.Add(1)
go func(index int) {
defer wg.Done()
// 異步任務(wù)邏輯
}(i)
}
// 等待所有協(xié)程完成
wg.Wait()4.使用errgroup實(shí)現(xiàn)協(xié)程組的錯誤處理
如果想簡單獲取協(xié)程返回的錯誤，errgroup包很適合，errgroup包是Go語言標(biāo)準(zhǔn)庫中的一個實(shí)用工具，用于管理一組協(xié)程并處理它們的錯誤?？梢允褂胑rrgroup.Group結(jié)構(gòu)來跟蹤和處理協(xié)程組的錯誤。
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 {
// 處理錯誤
}二、一些使用技巧
1.使用channel的range和close操作
range操作可以在接收通道上迭代值，直到通道關(guān)閉?？梢允褂胏lose函數(shù)關(guān)閉通道，以向接收方指示沒有更多的值。
ch := make(chan int)
go func() {
for i := 0; i < 5; i++ {
ch <- i // 發(fā)送值到通道
}
close(ch) // 關(guān)閉通道
}()
// 使用range迭代接收通道的值
for val := range ch {
// 處理接收到的值
}2.使用select語句實(shí)現(xiàn)多個異步操作的等待
ch1 := make(chan int)
ch2 := make(chan string)
go func() {
// 異步任務(wù)1邏輯
ch1 <- result1
}()
go func() {
// 異步任務(wù)2邏輯
ch2 <- result2
}()
// 在主goroutine中等待多個異步任務(wù)完成
select {
case res1 := <-ch1:
// 處理結(jié)果1
case res2 := <-ch2:
// 處理結(jié)果2
}3.使用select和time.After()實(shí)現(xiàn)超時(shí)控制
如果需要在異步操作中設(shè)置超時(shí)，可以使用select語句結(jié)合time.After()函數(shù)實(shí)現(xiàn)。
ch := make(chan int)
go func() {
// 異步任務(wù)邏輯
time.Sleep(2 * time.Second)
ch <- result
}()
// 設(shè)置超時(shí)時(shí)間
select {
case res := <-ch:
// 處理結(jié)果
case <-time.After(3 * time.Second):
// 超時(shí)處理
}4.使用select和time.After()實(shí)現(xiàn)超時(shí)控制
如果需要在異步操作中設(shè)置超時(shí)，可以使用select語句結(jié)合time.After()函數(shù)實(shí)現(xiàn)。
ch := make(chan int)
go func() {
// 異步任務(wù)邏輯
time.Sleep(2 * time.Second)
ch <- result
}()
// 設(shè)置超時(shí)時(shí)間
select {
case res := <-ch:
// 處理結(jié)果
case <-time.After(3 * time.Second):
// 超時(shí)處理
}5.使用time.Tick()和time.After()進(jìn)行定時(shí)操作
time.Tick()函數(shù)返回一個通道，定期發(fā)送時(shí)間值，可以用于執(zhí)行定時(shí)操作。time.After()函數(shù)返回一個通道，在指定的時(shí)間后發(fā)送一個時(shí)間值。
tick := time.Tick(1 * time.Second) // 每秒執(zhí)行一次操作
for {
select {
case <-tick:
// 執(zhí)行定時(shí)操作
}
}
select {
case <-time.After(5 * time.Second):
// 在5秒后執(zhí)行操作
}6.使用sync.Mutex或sync.RWMutex進(jìn)行并發(fā)安全訪問
當(dāng)多個協(xié)程并發(fā)訪問共享數(shù)據(jù)時(shí)，需要確保數(shù)據(jù)訪問的安全性。sync.Mutex和sync.RWMutex提供了互斥鎖和讀寫鎖，用于在訪問共享資源之前進(jìn)行鎖定，以避免數(shù)據(jù)競爭。sync.RWMutex是一種讀寫鎖，可以在多個協(xié)程之間提供對共享資源的并發(fā)訪問控制。多個協(xié)程可以同時(shí)獲取讀鎖，但只有一個協(xié)程可以獲取寫鎖。
var mutex sync.Mutex
var data int
// 寫操作，使用互斥鎖保護(hù)數(shù)據(jù)
mutex.Lock()
data = 123
mutex.Unlock()
// 讀操作，使用讀鎖保護(hù)數(shù)據(jù)
//RLock()加讀鎖時(shí)，如果存在寫鎖，則無法加讀鎖；當(dāng)只有讀鎖或者沒有鎖時(shí)，可以加讀鎖，讀鎖可以加載多個
mutex.RLock()
value := data
mutex.RUnlock()
var rwMutex sync.RWMutex
var sharedData map[string]string
// 讀操作，使用rwMutex.RLock讀鎖保護(hù)數(shù)據(jù)
func readData(key string) string {
rwMutex.RLock()
defer rwMutex.RUnlock()
return sharedData[key]
}
// 寫操作，使用rwMutex.Lock寫鎖保護(hù)數(shù)據(jù)
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()
7.使用sync.Cond進(jìn)行條件變量控制
sync.Cond是一個條件變量，用于在協(xié)程之間進(jìn)行通信和同步。它可以在指定的條件滿足之前阻塞等待，并在條件滿足時(shí)喚醒等待的協(xié)程。
var cond = sync.NewCond(&sync.Mutex{})
var ready bool
go func() {
// 異步任務(wù)邏輯
ready = true
// 通知等待的協(xié)程條件已滿足
cond.Broadcast()
}()
// 在某個地方等待條件滿足
cond.L.Lock()
for !ready {
cond.Wait()
}
cond.L.Unlock()8.使用sync.Pool管理對象池
sync.Pool是一個對象池，用于緩存和復(fù)用臨時(shí)對象，可以提高對象的分配和回收效率。
type MyObject struct {
// 對象結(jié)構(gòu)
}
var objectPool = sync.Pool{
New: func() interface{} {
// 創(chuàng)建新對象
return &MyObject{}
},
}
// 從對象池獲取對象
obj := objectPool.Get().(*MyObject)
// 使用對象
// 將對象放回對象池
objectPool.Put(obj)9.使用sync.Once實(shí)現(xiàn)只執(zhí)行一次的操作
sync.Once用于確保某個操作只執(zhí)行一次，無論有多少個協(xié)程嘗試執(zhí)行它，常用于初始化或加載資源等場景。
var once sync.Once
var resource *Resource
func getResource() *Resource {
once.Do(func() {
// 執(zhí)行初始化資源的操作，僅執(zhí)行一次
resource = initResource()
})
return resource
}
// 在多個協(xié)程中獲取資源
go func() {
res := getResource()
// 使用資源
}()
go func() {
res := getResource()
// 使用資源
}()10.使用sync.Once和context.Context實(shí)現(xiàn)資源清理
可以結(jié)合使用sync.Once和context.Context來確保在多個協(xié)程之間只執(zhí)行一次資源清理操作，并在取消或超時(shí)時(shí)進(jìn)行清理。
var once sync.Once
func cleanup() {
// 執(zhí)行資源清理操作
}
func doTask(ctx context.Context) {
go func() {
select {
case <-ctx.Done():
once.Do(cleanup) // 只執(zhí)行一次資源清理
}
}()
// 異步任務(wù)邏輯
}11.使用sync.Map實(shí)現(xiàn)并發(fā)安全的映射
sync.Map是Go語言標(biāo)準(zhǔn)庫中提供的并發(fā)安全的映射類型，可在多個協(xié)程之間安全地進(jìn)行讀寫操作。
var m sync.Map
// 存儲鍵值對
m.Store("key", "value")
// 獲取值
if val, ok := m.Load("key"); ok {
// 使用值
}
// 刪除鍵
m.Delete("key")12.使用context.Context進(jìn)行協(xié)程管理和取消
context.Context用于在協(xié)程之間傳遞上下文信息，并可用于取消或超時(shí)控制。可以使用context.WithCancel()創(chuàng)建一個可取消的上下文，并使用context.WithTimeout()創(chuàng)建一個帶有超時(shí)的上下文。
ctx, cancel := context.WithCancel(context.Background())
go func() {
// 異步任務(wù)邏輯
if someCondition {
cancel() // 取消任務(wù)
}
}()
// 等待任務(wù)完成或取消
select {
case <-ctx.Done():
// 任務(wù)被取消或超時(shí)
}13.使用context.WithDeadline()和context.WithTimeout()設(shè)置截止時(shí)間
context.WithDeadline()和context.WithTimeout()函數(shù)可以用于創(chuàng)建帶有截止時(shí)間的上下文，以限制異步任務(wù)的執(zhí)行時(shí)間。
func doTask(ctx context.Context) {
// 異步任務(wù)邏輯
select {
case <-time.After(5 * time.Second):
// 超時(shí)處理
case <-ctx.Done():
// 上下文取消處理
}
}
func main() {
ctx := context.Background()
ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
defer cancel()
go doTask(ctx)
// 繼續(xù)其他操作
}14.使用context.WithValue()傳遞上下文值
context.WithValue()函數(shù)可用于在上下文中傳遞鍵值對，以在協(xié)程之間共享和傳遞上下文相關(guān)的值。
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)
// 繼續(xù)其他操作
}15.使用atomic包進(jìn)行原子操作
atomic包提供了一組函數(shù)，用于實(shí)現(xiàn)原子操作，以確保在并發(fā)環(huán)境中對共享變量的讀寫操作是原子的。
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)
}
