我看到很多關(guān)于如何讓 Go 等待 x 個 goroutines 完成的教程和示例，但我想做的是確?？偸怯?x 個 goroutine 在運行，所以一旦一個 goroutine 結(jié)束就會啟動一個新的 goroutine .具體來說，我有幾十萬個“要做的事情”，它們正在處理一些來自 MySQL 的東西。所以它是這樣工作的：db, err := sql.Open("mysql", connection_string)checkErr(err)defer db.Close()rows,err := db.Query(`SELECT id FROM table`)checkErr(err)defer rows.Close()var id uintfor rows.Next() { err := rows.Scan(&id) checkErr(err) go processTheThing(id) }checkErr(err)rows.Close()目前，將推出數(shù)十萬線程processTheThing()。我需要的是最多啟動 x 個（我們稱之為 20 個）goroutines。因此，它首先為前 20 行啟動 20 個，從那時起，它將在當(dāng)前 goroutine 之一完成時為下一個 id 啟動一個新的 goroutine。所以在任何時間點總是有 20 個在運行。我敢肯定，這是很簡單/標(biāo)準(zhǔn)，但我似乎無法找到任何教程或例子或如何做到這一點的一個很好的解釋。
3 回答
郎朗坤
TA貢獻(xiàn)1921條經(jīng)驗 獲得超9個贊
您可能會發(fā)現(xiàn)Go Concurrency Patterns文章很有趣，尤其是有界并行性部分，它解釋了您需要的確切模式。
您可以使用空結(jié)構(gòu)的通道作為限制保護(hù)來控制并發(fā)工作程序 goroutine 的數(shù)量：
package main
import "fmt"
func main() {
maxGoroutines := 10
guard := make(chan struct{}, maxGoroutines)
for i := 0; i < 30; i++ {
guard <- struct{}{} // would block if guard channel is already filled
go func(n int) {
worker(n)
<-guard
}(i)
}
}
func worker(i int) { fmt.Println("doing work on", i) }
ABOUTYOU
TA貢獻(xiàn)1812條經(jīng)驗 獲得超5個贊
在這里，我認(rèn)為像這樣簡單的事情會起作用：
package main
import "fmt"
const MAX = 20
func main() {
sem := make(chan int, MAX)
for {
sem <- 1 // will block if there is MAX ints in sem
go func() {
fmt.Println("hello again, world")
<-sem // removes an int from sem, allowing another to proceed
}()
}
}
慕姐8265434
TA貢獻(xiàn)1813條經(jīng)驗 獲得超2個贊
感謝大家?guī)椭医鉀Q這個問題。但是，我認(rèn)為沒有人真正提供既有效又簡單/易懂的東西，盡管你們都幫助我理解了該技術(shù)。
我最后所做的是我認(rèn)為作為對我的具體問題的答案更容易理解和實用，所以我會在這里發(fā)布，以防其他人有同樣的問題。
不知何故，這最終看起來很像 OneOfOne 發(fā)布的內(nèi)容，這很棒，因為現(xiàn)在我明白了。但是 OneOfOne 的代碼一開始我發(fā)現(xiàn)很難理解，因為將函數(shù)傳遞給函數(shù)使得理解什么是什么非常令人困惑。我認(rèn)為這種方式更有意義：
package main
import (
"fmt"
"sync"
)
const xthreads = 5 // Total number of threads to use, excluding the main() thread
func doSomething(a int) {
fmt.Println("My job is",a)
return
}
func main() {
var ch = make(chan int, 50) // This number 50 can be anything as long as it's larger than xthreads
var wg sync.WaitGroup
// This starts xthreads number of goroutines that wait for something to do
wg.Add(xthreads)
for i:=0; i<xthreads; i++ {
go func() {
for {
a, ok := <-ch
if !ok { // if there is nothing to do and the channel has been closed then end the goroutine
wg.Done()
return
}
doSomething(a) // do the thing
}
}()
}
// Now the jobs can be added to the channel, which is used as a queue
for i:=0; i<50; i++ {
ch <- i // add i to the queue
}
close(ch) // This tells the goroutines there's nothing else to do
wg.Wait() // Wait for the threads to finish
}
- 3 回答
- 0 關(guān)注
- 228 瀏覽
添加回答
舉報
0/150
提交
取消