快捷導(dǎo)航

golang 限制同一時(shí)間的并發(fā)量操作

更新時(shí)間：2020年12月14日 15:27:48 作者：小辣抓

這篇文章主要介紹了golang 限制同一時(shí)間的并發(fā)量操作，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧

go的并發(fā)量是很厲害的，goroutine創(chuàng)建的代價(jià)極小，其中一個(gè)重要的原因是因?yàn)間o采用了分段棧技術(shù)，每一個(gè)goroutine只占極小的空間。與此同時(shí)，goroutine是語(yǔ)言層面的，減少了內(nèi)核態(tài)到用戶(hù)態(tài)的切換開(kāi)銷(xiāo)，并且goroutine摒棄了一些golang用不到的一些os thread的系統(tǒng)調(diào)用，創(chuàng)建代價(jià)小。

我們可以一瞬間創(chuàng)建很多個(gè)goroutine，這是相當(dāng)容易的。

乍一看，這與題目完全不符，前面說(shuō)了那么多，難道不是鼓勵(lì)我們多創(chuàng)建goroutine嗎？不不不，goroutine確實(shí)很好用，但是如果不加以限制，很有可能出現(xiàn)其他的不可預(yù)料的錯(cuò)誤。

比如在web領(lǐng)域中，一個(gè)連接，在linux/unix下就相當(dāng)于是打開(kāi)了一個(gè)文件，占用一個(gè)文件描述符。但是系統(tǒng)會(huì)規(guī)定文件描述符的上限，我們可以使用ulimit -n來(lái)進(jìn)行查看，如果我們遵循量大就好的話(huà)，那么一擁而上的請(qǐng)求連接會(huì)瞬間報(bào)錯(cuò)。

2018/06/30 10:09:54 dial tcp :8080: socket: too many open files

上面這條報(bào)錯(cuò)信息源于我寫(xiě)的一個(gè)循環(huán)請(qǐng)求的工具

package main
import (
  "sync"
  "net"
  "strconv"
  "fmt"
  "log"
)
const (
  MAX_CONCURRENCY = 10000 
)
var waitGroup sync.WaitGroup
func main(){
  concurrency()
  waitGroup.Wait()
}
//進(jìn)行網(wǎng)絡(luò)io
func request(currentCount int){
  fmt.Println("request" + strconv.Itoa(currentCount) + "\r")
  conn, err := net.Dial("tcp",":8080")
  if err != nil { log.Fatal(err) }
  defer conn.Close()
  defer waitGroup.Done()
}
//并發(fā)請(qǐng)求
func concurrency(){
  for i := 0;i < MAX_CONCURRENCY;i++ {
    waitGroup.Add(1)
    go request(i)
  }
}

用go建立一個(gè)服務(wù)端很簡(jiǎn)單，我這里簡(jiǎn)單的貼下server的代碼

package main
import (
  "io"
  "os"
  "fmt"
  "net"
)
func checkErr(err error){
  if err != nil { fmt.Fprintln(os.Stderr, err) }
}
func main() {
  listener, err := net.Listen("tcp",":8080")
  checkErr(err)
  for {
    conn, err := listener.Accept()
    checkErr(err)
    go func(conn net.Conn){ 
      _, err := io.WriteString(conn, "welcome!") 
      checkErr(err)
      defer conn.Close()
    }(conn)
  }
}

現(xiàn)在回到主題，我們可以看到一擁而上其實(shí)也有壞處，想要解決這一問(wèn)題，我們可以限制同一時(shí)間的并發(fā)數(shù)量，可以利用channel來(lái)達(dá)到這一點(diǎn)，這有點(diǎn)類(lèi)似于信號(hào)量（Semaphore）

創(chuàng)建一個(gè)帶緩存的channel，其中CHANNEL_CACHE為同一時(shí)間的最大并發(fā)量

想簡(jiǎn)單的說(shuō)一下為什么這里chan的類(lèi)型要用一個(gè)空的struct，這是因?yàn)樵谶@個(gè)場(chǎng)景下（限制同一時(shí)間的并發(fā)量），通過(guò)channel傳輸?shù)臄?shù)據(jù)的類(lèi)型并不重要，我們只需要通過(guò)做一個(gè)通知效果就行了（就像你通知你朋友起床，你只用閃個(gè)電話(huà)，而不用實(shí)際的接通，省去了電話(huà)費(fèi)的開(kāi)銷(xiāo)），這里的空的struct實(shí)際上是不占任何空間的，因此這里選用空的struct

const (
  CHANNEL_CACHE = 200
)
var tmpChannel = make(chan struct{}, CHANNEL_CACHE)

在與服務(wù)器建立連接的地方這樣寫(xiě)（是不是很類(lèi)似于信號(hào)量）

tmpChan <- struct{}{}
conn, err := net.Dial("tcp",":8080")
<- tmpChan

這樣同一時(shí)間的并發(fā)量就由CHANNEL_CACHE限制下來(lái)

經(jīng)過(guò)循環(huán)開(kāi)啟的goroutine在請(qǐng)求服務(wù)器之前會(huì)向channel發(fā)送消息，如果緩存滿(mǎn)了，那么說(shuō)明已經(jīng)有CHANNEL_CACHE個(gè)goroutine在進(jìn)行與服務(wù)器的連接，接著就會(huì)阻塞在這里，等待其中一個(gè)goroutine處理完之后，從channel中讀出一個(gè)空的struct，這時(shí)阻塞的地方向channel發(fā)送一個(gè)空struct，就可以與服務(wù)器建立連接了

下面貼一下全部的代碼

package main
import (
  "sync"
  "net"
  "strconv"
  "fmt"
  "log"
)
const (
  MAX_CONCURRENCY = 10000 
  CHANNEL_CACHE = 200
)
var tmpChan = make(chan struct{}, MAX_CONCURRENCY)
var waitGroup sync.WaitGroup
func main(){
  concurrency()
  waitGroup.Wait()
}
//進(jìn)行網(wǎng)絡(luò)io
func request(currentCount int){
  fmt.Println("request" + strconv.Itoa(currentCount) + "\r")
  tmpChan <- struct{}{}
  conn, err := net.Dial("tcp",":8080")
  <- tmpChan
  if err != nil { log.Fatal(err) }
  defer conn.Close()
  defer waitGroup.Done()
}
//并發(fā)
func concurrency(){
  for i := 0;i < MAX_CONCURRENCY;i++ {
    waitGroup.Add(1)
    go request(i)
	}
}

這樣就可以愉快的進(jìn)行并發(fā)了！?。?/p>

補(bǔ)充：Golang限制N個(gè)并發(fā)同時(shí)運(yùn)行

我就廢話(huà)不多說(shuō)了，大家還是直接看代碼吧~

package main 
import (
  "fmt"
  "sync"
  "time"
) 
var wg sync.WaitGroup 
func main() {
  var wg sync.WaitGroup
 
  sem := make(chan struct{}, 2) // 最多允許2個(gè)并發(fā)同時(shí)執(zhí)行
  taskNum := 10
 
  for i := 0; i < taskNum; i++ {
    wg.Add(1)
 
    go func(id int) {
      defer wg.Done()
 
      sem <- struct{}{}    // 獲取信號(hào)
      defer func() { <-sem }() // 釋放信號(hào)
 
      // do something for task
      time.Sleep(time.Second * 2)
      fmt.Println(id, time.Now())
    }(i)
  }
  wg.Wait()
}

以上為個(gè)人經(jīng)驗(yàn)，希望能給大家一個(gè)參考，也希望大家多多支持腳本之家。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教。

您可能感興趣的文章: