我正在嘗試同時運行一個函數。它調用我的數據庫可能需要 2-10 秒。我希望它完成后繼續執行下一個例程，即使另一個例程仍在處理中，但只希望它一次最多處理 2 個。我希望這種情況無限期地發生。我覺得我快到了，但是 waitGroup 強制這兩個例程等到完成，然后再繼續另一個迭代。const ROUTINES = 2;for { var wg sync.WaitGroup _, err:= db.Exec(`Random DB Call`) if err != nil { panic(err) } ch := createRoutines(db, &wg) wg.Add(ROUTINES) for i := 1; i <= ROUTINES; i++ { ch <- i time.Sleep(2 * time.Second) } close(ch) wg.Wait() }func createRoutines(db *sqlx.DB, wg *sync.WaitGroup) chan int { var ch = make(chan int, 5) for i := 0; i < ROUTINES ; i++ { go func(db *sqlx.DB) { defer wg.Done() for { _, ok := <-ch if !ok { return } doStuff(db) } }(db) } return ch}
2 回答
有只小跳蛙
TA貢獻1824條經驗 獲得超8個贊
如果您只需要同時運行 n 個 goroutine，您可以擁有一個大小為 n 的緩沖通道，并在沒有剩余空間時使用它來阻止創建新的 goroutine，就像這樣
package main
import (
"fmt"
"math/rand"
"time"
)
func main() {
const ROUTINES = 2
rand.Seed(time.Now().UnixNano())
stopper := make(chan struct{}, ROUTINES)
var counter int
for {
counter++
stopper <- struct{}{}
go func(c int) {
fmt.Println("+ Starting goroutine", c)
time.Sleep(time.Duration(rand.Intn(3)) * time.Second)
fmt.Println("- Stopping goroutine", c)
<-stopper
}(counter)
}
}
在這個例子中，你會看到你只能擁有 ROUTINES 個存活時間為 0、1 或 2 秒的 goroutine。在輸出中，您還可以看到每次一個 goroutine 結束時另一個 goroutine 是如何開始的。
慕姐8265434
TA貢獻1813條經驗 獲得超2個贊
這增加了一個外部依賴，但考慮這個實現：
package main
import (
"context"
"database/sql"
"log"
"github.com/MicahParks/ctxerrpool"
)
func main() {
// Create a pool of 2 workers for database queries. Log any errors.
databasePool := ctxerrpool.New(2, func(_ ctxerrpool.Pool, err error) {
log.Printf("Failed to execute database query.\nError: %s", err.Error())
})
// Get a list of queries to execute.
queries := []string{
"SELECT first_name, last_name FROM customers",
"SELECT price FROM inventory WHERE sku='1234'",
"other queries...",
}
// TODO Make a database connection.
var db *sql.DB
for _, query := range queries {
// Intentionally shadow the looped variable for scope.
query := query
// Perform the query on a worker. If no worker is ready, it will block until one is.
databasePool.AddWorkItem(context.TODO(), func(workCtx context.Context) (err error) {
_, err = db.ExecContext(workCtx, query)
return err
})
}
// Wait for all workers to finish.
databasePool.Wait()
}
- 2 回答
- 0 關注
- 209 瀏覽
添加回答
舉報
0/150
提交
取消