永远连续运行最多两个 goroutine

Question

我正在尝试同时运行一个函数。它调用我的数据库可能需要 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}

Answer 1

如果您只需要同时运行 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 是如何开始的。

Answer 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()

}