本篇內(nèi)容介紹了“Go中的并發(fā)方法實例代碼分析”的有關(guān)知識，在實際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓小編帶領(lǐng)大家學(xué)習(xí)一下如何處理這些情況吧！希望大家仔細(xì)閱讀，能夠?qū)W有所成！

城東網(wǎng)站制作公司哪家好，找創(chuàng)新互聯(lián)建站！從網(wǎng)頁設(shè)計、網(wǎng)站建設(shè)、微信開發(fā)、APP開發(fā)、自適應(yīng)網(wǎng)站建設(shè)等網(wǎng)站項目制作，到程序開發(fā)，運營維護。創(chuàng)新互聯(lián)建站自2013年創(chuàng)立以來到現(xiàn)在10年的時間，我們擁有了豐富的建站經(jīng)驗和運維經(jīng)驗，來保證我們的工作的順利進(jìn)行。專注于網(wǎng)站建設(shè)就選創(chuàng)新互聯(lián)建站。

Hello, Concurrent world

代碼很簡單——單個通道，單個goroutine，單次寫入，單次讀取。

package main

func main() {
    // 創(chuàng)建一個int類型的通道
    ch := make(chan int)

    // 開啟一個匿名 goroutine
    go func() {
        // 向通道發(fā)送數(shù)字42
        ch <- 42
    }()
    // 從通道中讀取
    <-ch
}

轉(zhuǎn)到交互式 WebGL 動畫

藍(lán)色線代表隨時間運行的goroutine. 連接‘main’和‘go #19’的藍(lán)色細(xì)線用來標(biāo)記goroutine的開始和結(jié)束同時展示了父子關(guān)系，最后，紅線代表發(fā)送/接收動作. 雖然這是兩個獨立的動作,我還是嘗試用“從 A 發(fā)送到 B”的動畫將他們表示成一個動作. goroutine 名稱中的“#19” 是 goroutine 真實的內(nèi)部ID, 其獲取方法參考了 Scott Mansfield 的 “Goroutine IDs” 這篇文章。

Timers

實際上，你可以通過以下方法構(gòu)建一個簡單的計時器——創(chuàng)建一個通道, 開啟一個 goroutine 讓其在指定的時間間隔后向通道中寫入數(shù)據(jù)，然后將這個通道返回給調(diào)用者。于是調(diào)用函數(shù)就會在讀取通道時阻塞，直到之前設(shè)定的時間間隔過去。接下來我們調(diào)用24次計時器然后嘗試具象化調(diào)用過程。

package main

import "time"

func timer(d time.Duration) <-chan int {
    c := make(chan int)
    go func() {
        time.Sleep(d)
        c <- 1
    }()
    return c
}

func main() {
    for i := 0; i < 24; i++ {
        c := timer(1 * time.Second)
        <-c
    }
}

轉(zhuǎn)到交互式 WebGL 動畫

很整潔，對嗎? 我們繼續(xù)。

Ping-pong

這個并發(fā)例子取自谷歌員工 Sameer Ajmani “Advanced Go Concurrency Patterns” 演講。當(dāng)然，這個模式不算非常高級，但是對于那些只熟悉Go的并發(fā)機制的人來說它看起來可能非常新鮮有趣。

這里我們用一個通道代表乒乓球臺. 一個整型變量代表球, 然后用兩個goroutine代表玩家，玩家通過增加整型變量的值（點擊計數(shù)器）模擬擊球動作。

package main

import "time"

func main() {
    var Ball int
    table := make(chan int)
    go player(table)
    go player(table)

    table <- Ball
    time.Sleep(1 * time.Second)
    <-table
}

func player(table chan int) {
    for {
        ball := <-table
        ball++
        time.Sleep(100 * time.Millisecond)
        table <- ball
    }
}

轉(zhuǎn)到交互式 WebGL 動畫

這里我建議你點擊 鏈接 進(jìn)入交互式 WebGL 動畫操作一下. 你可以放慢或者加速動畫，從不同的角度觀察。

現(xiàn)在，我們添加三個玩家看看。

    go player(table)
    go player(table)
    go player(table)

轉(zhuǎn)到交互式 WebGL 動畫 我們可以看到每個玩家都按照次序輪流操作，你可能會想為什么會這樣。為什么多個玩家（goroutine）會按照嚴(yán)格的順序接到“球”呢。

答案是 Go 運行時環(huán)境維護了一個 接收者 FIFO 隊列 (存儲需要從某一通道上接收數(shù)據(jù)的goroutine)，在我們的例子里，每個玩家在剛發(fā)出球后就做好了接球準(zhǔn)備。我們來看一下更復(fù)雜的情況，加入100個玩家。

for i := 0; i < 100; i++ {
    go player(table)
}

轉(zhuǎn)到交互式 WebGL 動畫

先進(jìn)先出順序很明顯了，是吧? 我們可以創(chuàng)建一百萬個goroutine，因為它們很輕量，但是對于實現(xiàn)我們的目的來說沒有必要。我們來想想其他可以玩的。 例如, 常見的消息傳遞模式。

Fan-In

并發(fā)世界中流行的模式之一是所謂的 fan-in 模式。這與 fan-out 模式相反，稍后我們將介紹。簡而言之，fan-in 是一項功能，可以從多個輸入中讀取數(shù)據(jù)并將其全部多路復(fù)用到單個通道中。

舉例來說：

package main

import (
    "fmt"
    "time"
)

func producer(ch chan int, d time.Duration) {
    var i int
    for {
        ch <- i
        i++
        time.Sleep(d)
    }
}

func reader(out chan int) {
    for x := range out {
        fmt.Println(x)
    }
}

func main() {
    ch := make(chan int)
    out := make(chan int)
    go producer(ch, 100*time.Millisecond)
    go producer(ch, 250*time.Millisecond)
    go reader(out)
    for i := range ch {
        out <- i
    }
}

Go to interactive WebGL animation

如我們所見，第一個 producer 每100毫秒生成一次值，第二個每250毫秒生成一次值，但是 reader 會立即從這兩個生產(chǎn)者那里接受值。實際上，多路復(fù)用發(fā)生在 main 的range循環(huán)中。

Workers

與 fan-in 相反的模式是 fan-out 或者worker 模式。多個 goroutine 可以從單個通道讀取，從而在CPU內(nèi)核之間分配大量的工作量，因此是 worker 的名稱。在Go中，此模式易于實現(xiàn)-只需以通道為參數(shù)啟動多個goroutine，然后將值發(fā)送至該通道-Go運行時會自動地進(jìn)行分配和復(fù)用 :)

package main

import (
    "fmt"
    "sync"
    "time"
)

func worker(tasksCh <-chan int, wg *sync.WaitGroup) {
    defer wg.Done()
    for {
        task, ok := <-tasksCh
        if !ok {
            return
        }
        d := time.Duration(task) * time.Millisecond
        time.Sleep(d)
        fmt.Println("processing task", task)
    }
}

func pool(wg *sync.WaitGroup, workers, tasks int) {
    tasksCh := make(chan int)

    for i := 0; i < workers; i++ {
        go worker(tasksCh, wg)
    }

    for i := 0; i < tasks; i++ {
        tasksCh <- i
    }

    close(tasksCh)
}

func main() {
    var wg sync.WaitGroup
    wg.Add(36)
    go pool(&wg, 36, 50)
    wg.Wait()
}

這里值得一提的是：并行性。如您所見，所有g(shù)oroutine并行’運行‘，等待通道給予它們’工作‘。鑒于上面的動畫，很容易發(fā)現(xiàn)goroutine幾乎立即接連地收到它們的工作。不幸的是，該動畫在goroutine確實在處理工作還是僅僅是在等待輸入的地方?jīng)]有用顏色顯示出來，但是此動畫是在GOMAXPROCS=4的情況下錄制的，因此只有4個goroutine有效地并行運行。我們將很快討論這個主題。

現(xiàn)在，讓我們做一些更復(fù)雜的事情，并啟動一些有自己workers（subworkers）的workers。

package main

import (
    "fmt"
    "sync"
    "time"
)

const (
    WORKERS    = 5
    SUBWORKERS = 3
    TASKS      = 20
    SUBTASKS   = 10
)

func subworker(subtasks chan int) {
    for {
        task, ok := <-subtasks
        if !ok {
            return
        }
        time.Sleep(time.Duration(task) * time.Millisecond)
        fmt.Println(task)
    }
}

func worker(tasks <-chan int, wg *sync.WaitGroup) {
    defer wg.Done()
    for {
        task, ok := <-tasks
        if !ok {
            return
        }

        subtasks := make(chan int)
        for i := 0; i < SUBWORKERS; i++ {
            go subworker(subtasks)
        }
        for i := 0; i < SUBTASKS; i++ {
            task1 := task * i
            subtasks <- task1
        }
        close(subtasks)
    }
}

func main() {
    var wg sync.WaitGroup
    wg.Add(WORKERS)
    tasks := make(chan int)

    for i := 0; i < WORKERS; i++ {
        go worker(tasks, &wg)
    }

    for i := 0; i < TASKS; i++ {
        tasks <- i
    }

    close(tasks)
    wg.Wait()
}

Go to interactive WebGL animation 很好。當(dāng)然，我們可以將worker和subworker的數(shù)量設(shè)置為更高的值，但是我試圖使動畫清晰易懂。

更酷的 fan-out 模式確實存在，例如動態(tài)數(shù)量的worker/subworker，通過通道發(fā)送通道，但是 fan-out 的想法現(xiàn)在應(yīng)該很清楚了。

服務(wù)器

下一個常見的模式類似于扇出，但是會在很短的時間內(nèi)生成goroutine，只是為了完成某些任務(wù)。它通常用于實現(xiàn)服務(wù)器-創(chuàng)建偵聽器，循環(huán)運行accept()并為每個接受的連接啟動goroutine。它非常具有表現(xiàn)力，可以實現(xiàn)盡可能簡單的服務(wù)器處理程序?？匆粋€簡單的例子：

package main

import "net"

func handler(c net.Conn) {
    c.Write([]byte("ok"))
    c.Close()
}

func main() {
    l, err := net.Listen("tcp", ":5000")
    if err != nil {
        panic(err)
    }
    for {
        c, err := l.Accept()
        if err != nil {
            continue
        }
        go handler(c)
    }
}

Go to 交互式WebGL動畫

這不是很有趣-似乎并發(fā)方面沒有發(fā)生任何事情。當(dāng)然，在引擎蓋下有很多復(fù)雜性，這是我們特意隱藏的。 “簡單性很復(fù)雜”.

但是，讓我們回到并發(fā)性并向我們的服務(wù)器添加一些交互。假設(shè)每個處理程序都希望異步寫入記錄器。在我們的示例中，記錄器本身是一個單獨的goroutine，它可以完成此任務(wù)。

package main

import (
    "fmt"
    "net"
    "time"
)

func handler(c net.Conn, ch chan string) {
    ch <- c.RemoteAddr().String()
    c.Write([]byte("ok"))
    c.Close()
}

func logger(ch chan string) {
    for {
        fmt.Println(<-ch)
    }
}

func server(l net.Listener, ch chan string) {
    for {
        c, err := l.Accept()
        if err != nil {
            continue
        }
        go handler(c, ch)
    }
}

func main() {
    l, err := net.Listen("tcp", ":5000")
    if err != nil {
        panic(err)
    }
    ch := make(chan string)
    go logger(ch)
    go server(l, ch)
    time.Sleep(10 * time.Second)
}

Go to 交互式WebGL動畫

不是嗎？但是很容易看到，如果請求數(shù)量增加并且日志記錄操作花費一些時間(例如，準(zhǔn)備和編碼數(shù)據(jù))，我們的* logger * goroutine很快就會成為瓶頸。我們可以使用一個已知的扇出模式。我們開始做吧。

服務(wù)器+工作者

帶工作程序的服務(wù)器示例是記錄器的高級版本。它不僅可以完成一些工作，而且還可以通過* results *通道將其工作結(jié)果發(fā)送回池中。沒什么大不了的，但是它將我們的記錄器示例擴展到了更實際的示例。

讓我們看一下代碼和動畫：

package main

import (
    "net"
    "time"
)

func handler(c net.Conn, ch chan string) {
    addr := c.RemoteAddr().String()
    ch <- addr
    time.Sleep(100 * time.Millisecond)
    c.Write([]byte("ok"))
    c.Close()
}

func logger(wch chan int, results chan int) {
    for {
        data := <-wch
        data++
        results <- data
    }
}

func parse(results chan int) {
    for {
        <-results
    }
}

func pool(ch chan string, n int) {
    wch := make(chan int)
    results := make(chan int)
    for i := 0; i < n; i++ {
        go logger(wch, results)
    }
    go parse(results)
    for {
        addr := <-ch
        l := len(addr)
        wch <- l
    }
}

func server(l net.Listener, ch chan string) {
    for {
        c, err := l.Accept()
        if err != nil {
            continue
        }
        go handler(c, ch)
    }
}

func main() {
    l, err := net.Listen("tcp", ":5000")
    if err != nil {
        panic(err)
    }
    ch := make(chan string)
    go pool(ch, 4)
    go server(l, ch)
    time.Sleep(10 * time.Second)
}

Go to 交互式WebGL動畫 我們在4個goroutine之間分配了工作，有效地提高了記錄器的吞吐量，但是從此動畫中，我們可以看到記錄器仍然可能是問題的根源。成千上萬的連接在分配之前會匯聚在一個通道中，這可能導(dǎo)致記錄器再次成為瓶頸。但是，當(dāng)然，它會在更高的負(fù)載下發(fā)生。

并發(fā)素篩(素篩指素數(shù)篩法)

足夠的扇入/扇出樂趣。讓我們看看更復(fù)雜的并發(fā)算法。我最喜歡的例子之一是Concurrent Prime Sieve，可以在[Go Concurrency Patterns]對話中找到。素數(shù)篩，或[Eratosthenes篩)是一種古老的算法，用于查找達(dá)到給定限制的素數(shù)。它通過按順序消除所有質(zhì)數(shù)的倍數(shù)來工作。天真的算法并不是真正有效的算法，尤其是在多核計算機上。

該算法的并發(fā)變體使用goroutine過濾數(shù)字-每個發(fā)現(xiàn)的素數(shù)一個goroutine，以及用于將數(shù)字從生成器發(fā)送到過濾器的通道。找到質(zhì)數(shù)后，它將通過通道發(fā)送到* main *以進(jìn)行輸出。當(dāng)然，該算法也不是很有效，特別是如果您想找到大質(zhì)數(shù)并尋找最低的Big O復(fù)雜度，但是我發(fā)現(xiàn)它非常優(yōu)雅。

// 并發(fā)的主篩
package main

import "fmt"

// 將序列2、3、4，...發(fā)送到頻道“ ch”。
func Generate(ch chan<- int) {
    for i := 2; ; i++ {
        ch <- i // Send 'i' to channel 'ch'.
    }
}

//將值從通道“ in”復(fù)制到通道“ out”，
//刪除可被“素數(shù)”整除的那些。
func Filter(in <-chan int, out chan<- int, prime int) {
    for {
        i := <-in // Receive value from 'in'.
        if i%prime != 0 {
            out <- i // Send 'i' to 'out'.
        }
    }
}

//主篩：菊花鏈過濾器過程。
func main() {
    ch := make(chan int) // Create a new channel.
    go Generate(ch)      // Launch Generate goroutine.
    for i := 0; i < 10; i++ {
        prime := <-ch
        fmt.Println(prime)
        ch2 := make(chan int)
        go Filter(ch, ch2, prime)
        ch = ch2
    }
}

轉(zhuǎn)到交互式WebGL動畫

，請以交互模式隨意播放此動畫。我喜歡它的說明性-它確實可以幫助您更好地理解該算法。 * generate * goroutine發(fā)出從2開始的每個整數(shù)，每個新的goroutine僅過濾特定的質(zhì)數(shù)倍數(shù)-2、3、5、7 …，將第一個找到的質(zhì)數(shù)發(fā)送給* main *。如果旋轉(zhuǎn)它從頂部看，您會看到從goroutine發(fā)送到main的所有數(shù)字都是質(zhì)數(shù)。漂亮的算法，尤其是在3D中。

GOMAXPROCS（調(diào)整并發(fā)的運行性能）

現(xiàn)在，讓我們回到我們的工作人員示例。還記得我告訴過它以GOMAXPROCS = 4運行嗎？那是因為所有這些動畫都不是藝術(shù)品，它們是真實程序的真實痕跡。

讓我們回顧一下GOMAXPROCS是什么。

GOMAXPROCS設(shè)置可以同時執(zhí)行的最大CPU數(shù)量。

當(dāng)然，CPU是指邏輯CPU。我修改了一些示例，以使他們真正地工作(而不僅僅是睡覺)并使用實際的CPU時間。然后，我運行了代碼，沒有進(jìn)行任何修改，只是設(shè)置了不同的GOMAXPROCS值。 Linux機頂盒有2個CPU，每個CPU具有12個內(nèi)核，因此有24個內(nèi)核。

因此，第一次運行演示了該程序在1個內(nèi)核上運行，而第二次-使用了所有24個內(nèi)核的功能。

WebGL動畫-1| WebGL動畫-24GOMAXPROCS1

這些動畫中的時間速度是不同的(我希望所有動畫都適合同一時間/ height)，因此區(qū)別很明顯。當(dāng)GOMAXPROCS = 1時，下一個工作人員只有在上一個工作完成后才能開始實際工作。在GOMAXPROCS = 24的情況下，加速非常大，而復(fù)用的開銷可以忽略不計。

不過，重要的是要了解，增加GOMAXPROCS并不總是可以提高性能，在某些情況下實際上會使它變得更糟。

Goroutines leak

我們可以從Go中的并發(fā)時間中證明什么呢？我想到的一件事情是goroutine泄漏。例如，如果您啟動goroutine，但超出范圍，可能會發(fā)生泄漏?；蛘?，您只是忘記添加結(jié)束條件，而運行了for{}循環(huán)。

第一次在代碼中遇到goroutine泄漏時，我的腦海中出現(xiàn)了可怕的圖像，并且在下個周末我寫了 expvarmon?，F(xiàn)在，我可以使用WebGL可視化該恐怖圖像。

看一看：

僅僅是看到此，我都會感到痛苦:) 所有這些行都浪費了資源，并且是您程序的定時炸彈。

Parallelism is not Concurrency

我要說明的最后一件事是并行性與并發(fā)性之間的區(qū)別。這個話題涵蓋了   很多 ，Rob Pike在這個話題上做了一個精彩的演講。確實是#必須觀看的視頻之一。

簡而言之，

并行是簡單的并行運行事物。

并發(fā)是一種構(gòu)造程序的方法。

因此，并發(fā)程序可能是并行的，也可能不是并行的，這些概念在某種程度上是正交的。我們在演示 GOMAXPROCS 設(shè)置效果時已經(jīng)看到了這一點。

我可以重復(fù)所有這些鏈接的文章和談話，但是一張圖片相當(dāng)于說了一千個字。我在這里能做的是可視化這個差異。因此，這是并行。許多事情并行運行。

轉(zhuǎn)到交互式WebGL動畫

這也是并行性：

轉(zhuǎn)到交互式WebGL動畫

但這是并發(fā)的：

還有這個：

這也是并發(fā)的：

How it was made

為了創(chuàng)建這些動畫，我編寫了兩個程序：gotracer 和 gothree.js 庫。首先，gotracer執(zhí)行以下操作：

• 解析Go程序的AST樹（Abstract Syntax Tree，抽象語法樹），并在與并發(fā)相關(guān)的事件上插入帶有輸出的特殊命令-啟動/停止goroutine，創(chuàng)建通道，向/從通道發(fā)送/接收。

• 運行生成的程序

• 分析此特殊輸出，并生成帶有事件和時間戳描述的JSON。

生成的JSON示例：

“Go中的并發(fā)方法實例代碼分析”的內(nèi)容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業(yè)相關(guān)的知識可以關(guān)注創(chuàng)新互聯(lián)網(wǎng)站，小編將為大家輸出更多高質(zhì)量的實用文章！