快捷導(dǎo)航

Go并發(fā)編程實(shí)踐

更新時(shí)間：2017年01月24日 22:24:53 投稿：mdxy-dxy

并發(fā)編程一直是Golang區(qū)別與其他語(yǔ)言的很大優(yōu)勢(shì)，也是實(shí)際工作場(chǎng)景中經(jīng)常遇到的。近日筆者在組內(nèi)分享了我們常見(jiàn)的并發(fā)場(chǎng)景，及代碼示例，以期望大家能在遇到相同場(chǎng)景下，能快速的想到解決方案，或者是拿這些方案與自己實(shí)現(xiàn)的比較，取長(zhǎng)補(bǔ)短?，F(xiàn)整理出來(lái)與大家共享

前言

并發(fā)編程一直是Golang區(qū)別與其他語(yǔ)言的很大優(yōu)勢(shì)，也是實(shí)際工作場(chǎng)景中經(jīng)常遇到的。近日筆者在組內(nèi)分享了我們常見(jiàn)的并發(fā)場(chǎng)景，及代碼示例，以期望大家能在遇到相同場(chǎng)景下，能快速的想到解決方案，或者是拿這些方案與自己實(shí)現(xiàn)的比較，取長(zhǎng)補(bǔ)短?，F(xiàn)整理出來(lái)與大家共享。

簡(jiǎn)單并發(fā)場(chǎng)景

很多時(shí)候，我們只想并發(fā)的做一件事情，比如測(cè)試某個(gè)接口的是否支持并發(fā)。那么我們就可以這么做：

func RunScenario1() {
    count := 10
    var wg sync.WaitGroup

    for i := 0; i < count; i++ {
       wg.Add(1)
       go func(index int) {
           defer wg.Done()
           doSomething(index)
       }(i)
    }

    wg.Wait()
}

使用goroutine來(lái)實(shí)現(xiàn)異步，使用WaitGroup來(lái)等待所有g(shù)oroutine結(jié)束。這里要注意的是要正確釋放WaitGroup的counter(在goroutine里調(diào)用Done()方法)。

但此種方式有個(gè)弊端，就是當(dāng)goroutine的量過(guò)多時(shí)，很容易消耗完客戶端的資源，導(dǎo)致程序表現(xiàn)不佳。

規(guī)定時(shí)間內(nèi)的持續(xù)并發(fā)模型

我們?nèi)匀灰詼y(cè)試某個(gè)后端API接口為例，如果我們想知道這個(gè)接口在持續(xù)高并發(fā)情況下是否有句柄泄露，這種情況該如何測(cè)試呢？

這種時(shí)候，我們需要能控制時(shí)間的高并發(fā)模型：

func RunScenario2() {
  timeout := time.Now().Add(time.Second * time.Duration(10))
  n := runtime.NumCPU()

  waitForAll := make(chan struct{})
  done := make(chan struct{})
  concurrentCount := make(chan struct{}, n)

  for i := 0; i < n; i++ {
    concurrentCount <- struct{}{}
  }

  go func() {
    for time.Now().Before(timeout) {
      <-done
      concurrentCount <- struct{}{}
    }

    waitForAll <- struct{}{}
  }()

  go func() {
    for {
      <-concurrentCount
      go func() {
        doSomething(rand.Intn(n))
        done <- struct{}{}
      }()
    }
  }()

  <-waitForAll
}

上面的代碼里，我們通過(guò)一個(gè)buffered channel來(lái)控制并發(fā)的數(shù)量(concurrentCount)，然后另起一個(gè)channel來(lái)周期性的發(fā)起新的任務(wù)，而控制的條件就是 time.Now().Before(timeout)，這樣當(dāng)超過(guò)規(guī)定的時(shí)間，waitForAll 就會(huì)得到信號(hào)，而使整個(gè)程序退出。

這是一種實(shí)現(xiàn)方式，那么還有其他的方式?jīng)]？我們接著往下看。

基于大數(shù)據(jù)量的并發(fā)模型

前面說(shuō)的基于時(shí)間的并發(fā)模型，那如果只知道數(shù)據(jù)量很大，但是具體結(jié)束時(shí)間不確定，該怎么辦呢？

比如，客戶給了個(gè)幾TB的文件列表，要求把這些文件從存儲(chǔ)里刪除。再比如，實(shí)現(xiàn)個(gè)爬蟲去爬某些網(wǎng)站的所有內(nèi)容。

而解決此類問(wèn)題，最常見(jiàn)的就是使用工作池模式了(Worker Pool)。以刪文件為例，我們可以簡(jiǎn)單這樣來(lái)處理：

Jobs - 可以從文件列表里讀取文件，初始化為任務(wù)，然后發(fā)給worker
Worker - 拿到任務(wù)開(kāi)始做事
Collector - 收集worker處理后的結(jié)果
Worker Pool - 控制并發(fā)的數(shù)量

雖然這只是個(gè)簡(jiǎn)單Worker Pool模型，但已經(jīng)能滿足我們的需求：

func RunScenario3() {
    numOfConcurrency := runtime.NumCPU()
    taskTool := 10
    jobs := make(chan int, taskTool)
    results := make(chan int, taskTool)
    var wg sync.WaitGroup

    // workExample
    workExampleFunc := func(id int, jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {
       defer wg.Done()
       for job := range jobs {
           res := job * 2
           fmt.Printf("Worker %d do things, produce result %d \n", id, res)
           time.Sleep(time.Millisecond * time.Duration(100))
           results <- res
       }
    }

    for i := 0; i < numOfConcurrency; i++ {
       wg.Add(1)
       go workExampleFunc(i, jobs, results, &wg)
    }

    totalTasks := 100 // 本例就要從文件列表里讀取

    wg.Add(1)
    go func() {
       defer wg.Done()
       for i := 0; i < totalTasks; i++ {
           n := <-results
           fmt.Printf("Got results %d \n", n)
       }
       close(results)
    }()

    for i := 0; i < totalTasks; i++ {
       jobs <- i
    }
    close(jobs)
    wg.Wait()
}

在Go里，分發(fā)任務(wù)，收集結(jié)果，我們可以都交給Channel來(lái)實(shí)現(xiàn)。從實(shí)現(xiàn)上更加的簡(jiǎn)潔。

仔細(xì)看會(huì)發(fā)現(xiàn)，本模型也是適用于按時(shí)間來(lái)控制并發(fā)。只要把totalTask的遍歷換成時(shí)間控制就好了。

等待異步任務(wù)執(zhí)行結(jié)果

goroutine和channel的組合在實(shí)際編程時(shí)經(jīng)常會(huì)用到，而加上Select更是無(wú)往而不利。

func RunScenario4() {
    sth := make(chan string)
    result := make(chan string)
    go func() {
       id := rand.Intn(100)
       for {
           sth <- doSomething(id)
       }
    }()
    go func() {
       for {
           result <- takeSomthing(<-sth)
       }
    }()

    select {
    case c := <-result:
       fmt.Printf("Got result %s ", c)
    case <-time.After(time.Duration(30 * time.Second)):
       fmt.Errorf("指定時(shí)間內(nèi)都沒(méi)有得到結(jié)果")
    }
}

在select的case情況，加上time.After()模型可以讓我們?cè)谝欢〞r(shí)間范圍內(nèi)等待異步任務(wù)結(jié)果，防止程序卡死。

定時(shí)反饋異步任務(wù)結(jié)果

上面我們說(shuō)到持續(xù)的壓測(cè)某后端API，但并未實(shí)時(shí)收集結(jié)果。而很多時(shí)候?qū)τ谛阅軠y(cè)試場(chǎng)景，實(shí)時(shí)的統(tǒng)計(jì)吞吐率，成功率是非常有必要的。

func RunScenario5() {
  concurrencyCount := runtime.NumCPU()
  for i := 0; i < concurrencyCount; i++ {
    go func(index int) {
      for {
        doUploadMock()
      }
    }(i)
  }

  t := time.NewTicker(time.Second)
  for {
    select {
    case <-t.C:
      // 計(jì)算并打印實(shí)時(shí)數(shù)據(jù)
    }
  } 
}

這種場(chǎng)景就需要使用到Ticker，且上面的Example模型還能控制并發(fā)數(shù)量，也是非常實(shí)用的方式。

知識(shí)點(diǎn)總結(jié)

上面我們共提到了五種并發(fā)模式：