快捷導(dǎo)航

golang并發(fā)執(zhí)行的幾種方式小結(jié)

更新時間：2023年08月23日 11:07:03 作者：阿十~

本文主要介紹了golang并發(fā)執(zhí)行的幾種方式小結(jié)，主要包括了Channel，WaitGroup ，Context，使用這三種機制中的一種或者多種可以達(dá)到并發(fā)控制很好的效果，具有一定的參考價值，感興趣的可以了解一下

背景

主要記錄一下工作中和各個文檔里關(guān)于golang并發(fā)開發(fā)的實踐。golang并發(fā)主要用到了

Channel: 使用channel控制子協(xié)程，WaitGroup : 使用信號量機制控制子協(xié)程，Context: 使用上下文控制子協(xié)程。使用這三種機制中的一種或者多種可以達(dá)到并發(fā)控制很好的效果。關(guān)于這三個知識點，http://m.fzitv.net/jiaoben/296040z3m.htm介紹的比較詳細(xì)了，這里只介紹幾個場景用法。

實踐

waitGroup并發(fā)執(zhí)行不同任務(wù)

這種寫法適合并發(fā)處理少量case，并且funcA和funcB的作用是不同任務(wù)的時候。

? ? wg := sync.WaitGroup{}
? ? var result1 interface{}
    var result2 interface{}
? ? wg.Add(2)
    go func() {
        defer func() {
            wg.Done()
        }()
        result1 = funcA()
    }()
? ? go func() {
        defer func() {
            wg.Done()
        }()
        result1 = funcB()
    }()  
    wg.wait()

waitGroup并發(fā)執(zhí)行相同任務(wù)

假設(shè)有一批url，需要并發(fā)去抓取，這個時候可能只是請求的地址不同，任務(wù)的函數(shù)是一致的。這時候可以使用for循環(huán)的方式去批量執(zhí)行。使用該方法時候，需要使用線程安全的結(jié)構(gòu)去做數(shù)據(jù)同步。此外下文的寫法最大的弊端是沒法做并發(fā)度控制，如果請求過多，容易把下游打滿，以及啟過多協(xié)程，浪費資源，只適合小數(shù)據(jù)集，

import (
    "fmt"
    "io/ioutil"
    "net/http"
    "strings"
    "sync"
)
func main() {
    idList := []string{"x", "xx"}
    wg := sync.WaitGroup{}
    dataMap := sync.Map{} //  sync.Map是線程安全的，如果使用 map去存儲返回結(jié)果會報錯，
                          //  接受返回結(jié)果也可以是channel，channel也是線程安全的
    for _, id := range idList {
        wg.Add(1)
        go func(id string) {
            defer func() {
                wg.Done()
            }()
            data := PostData(id)
            dataMap.Store(id, data)
        }(id)
    }
    wg.Wait()
    for _, id := range idList {
        fmt.Println(dataMap.Load(id))
    }
}
func PostData(id string) string {
    url := "http:xx"
    method := "POST"
    payload := strings.NewReader(fmt.Sprintf(`{"id":"%s"}`, id))
    client := &http.Client{}
    req, err := http.NewRequest(method, url, payload)
    if err != nil {
        fmt.Println(err)
        return ""
    }
    req.Header.Add("Content-Type", "application/json")
    res, err := client.Do(req)
    if err != nil {
        fmt.Println(err)
        return ""
    }
    defer res.Body.Close()
    body, err := ioutil.ReadAll(res.Body)
    if err != nil {
        fmt.Println(err)
        return ""
    }
    return string(body)
}

加入channnel，控制并發(fā)度

package main
import (
    "fmt"
    "net/http"
    "sync"
)
func main() {
    urls := []string{"http://a.com", "http://b.com", "http://c.com"}
    // 控制并發(fā)度為2
    concurrency := 2
    sem := make(chan struct{}, concurrency)
    var wg sync.WaitGroup
    for _, url := range urls {
        wg.Add(1)
        go func(url string) {
            sem <- struct{}{} // 獲取信號量
            defer func() {
                <-sem // 釋放信號量
                wg.Done()
            }()
            resp, err := http.Get(url)
            if err != nil {
                fmt.Printf("Error fetching %s: %v\n", url, err)
                return
            }
            defer resp.Body.Close()
            fmt.Printf("Fetched %s with status code %d\n", url, resp.StatusCode)
        }(url)
    }
    wg.Wait()
    fmt.Println("All URLs fetched")
}

使用channel進(jìn)行并發(fā)編程

接下來，我們使用一個for循環(huán)來遍歷URL切片，并為每個URL啟動一個goroutine。在每個goroutine中，我們首先從并發(fā)度通道中獲取一個信號，表示可以開始請求。然后，我們發(fā)送一個http GET請求，并將響應(yīng)結(jié)果發(fā)送到結(jié)果通道中。最后，我們釋放一個信號，表示請求已完成。

在主函數(shù)中，我們使用另一個for循環(huán)從結(jié)果通道中讀取所有響應(yīng)結(jié)果，并將它們打印到控制臺上。

需要注意的是，我們在并發(fā)度通道中使用了一個空結(jié)構(gòu)體{}，因為我們只需要通道來控制并發(fā)度，而不需要在通道中傳遞任何數(shù)據(jù)。此外，我們還使用了通道的阻塞特性來控制并發(fā)度，因為當(dāng)通道已滿時，任何試圖向通道中發(fā)送數(shù)據(jù)的操作都會被阻塞，直到有空間可用為止。

package main
import (
    "fmt"
    "net/http"
)
func main() {
    urls := []string{"http://www.google.com", "http://www.facebook.com", "http://www.apple.com"}
    // 創(chuàng)建一個通道來控制并發(fā)度為2
    concurrency := make(chan struct{}, 2)
    // 創(chuàng)建一個通道來接收響應(yīng)結(jié)果
    results := make(chan string, len(urls))
    for _, url := range urls {
        // 啟動一個goroutine來請求url
        go func(url string) {
            // 從通道中獲取一個信號，表示可以開始請求
            concurrency <- struct{}{}
            // 發(fā)送http GET請求
            resp, err := http.Get(url)
            if err != nil {
                results <- fmt.Sprintf("%s -> error: %s", url, err)
            } else {
                results <- fmt.Sprintf("%s -> status: %s", url, resp.Status)
                resp.Body.Close()
            }
            // 釋放一個信號，表示請求已完成
            <-concurrency
        }(url)
    }
    // 從結(jié)果通道中讀取所有響應(yīng)結(jié)果
    for i := 0; i < len(urls); i++ {
        fmt.Println(<-results)
    }
}

使用context，進(jìn)行并發(fā)控制

這種機制下生成的goruntine是樹形結(jié)構(gòu)的，有依賴關(guān)系。

func getData(ctx context.Context, result chan string, id string) {
    for {
        select {
        case <-ctx.Done():
            fmt.Println("running get Data")
            return
        default:
            resultData := PostData(id)
            result <- resultData
        }
    }
}
func main() {
    idList := []string{"xx", "xxx"}
    ctx := context.Background()
    var result = make(chan string, 2)
    go getData(ctx, result, idList[0])
    go getData(ctx, result, idList[1])
    fmt.Println(<-result)
    fmt.Println(<-result)
}
func PostData(id string) string {
    url := "http://xxx"
    method := "POST"
    payload := strings.NewReader(fmt.Sprintf(`{"id":"%s"}`, id))
    client := &http.Client{}
    req, err := http.NewRequest(method, url, payload)
    if err != nil {
        fmt.Println(err)
        return ""
    }
    req.Header.Add("Content-Type", "application/json")
    res, err := client.Do(req)
    if err != nil {
        fmt.Println(err)
        return ""
    }
    defer res.Body.Close()
    body, err := ioutil.ReadAll(res.Body)
    if err != nil {
        fmt.Println(err)
        return ""
    }
    return string(body)
}

到此這篇關(guān)于golang并發(fā)執(zhí)行的幾種方式小結(jié)的文章就介紹到這了,更多相關(guān)golang并發(fā)執(zhí)行內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: