用于处理查询的工作池答案

【问题标题】：Worker pool to handle queries用于处理查询的工作池
【发布时间】：2014-11-03 12:19:43
【问题描述】：

我对 Go 很陌生，正在寻找一种方法来使用 100 个工作人员处理 3000 个查询并确保每个工作人员都有一个连接（MySQL 已经配置了超过 100 个连接）。这是我的尝试：

package main

import (
    "database/sql"
    _ "github.com/go-sql-driver/mysql"
)

var query *sql.Stmt

func worker(jobs <-chan int, results chan<- int) {

    for _ = range jobs {

        _, e := query.Exec("a")

        if e != nil {

            panic(e.Error())
        }

        results <- 1
    }
}

func main() {

    workers := 100

    db, e := sql.Open("mysql", "foo:foo@/foo")

    if e != nil {

        panic(e.Error())
    }

    db.SetMaxOpenConns(workers)
    db.SetMaxIdleConns(workers)

    defer db.Close()

    query, e = db.Prepare("INSERT INTO foo (foo) values(?)")

    if e != nil {

        panic(e.Error())
    }

    total := 30000
    jobs := make(chan int, total)
    results := make(chan int, total)

    for w := 0; w < workers; w++ {

        go worker(jobs, results)
    }

    for j := 0; j < total; j++ {

        jobs <- j
    }

    close(jobs)

    for r := 0; r < total; r++ {

        <-results
    }
}

它正在工作，但我不确定是否是最好的方法。

如果您认为这是基于意见或根本不是一个好问题，请将其标记为已关闭并发表评论解释原因。

【问题讨论】：

"Concurrency is not parallelism" by Rob Pike 从第 19 分钟开始——“一个非常简单的负载均衡器”，可以在一些工作人员之间分配工作。
@tomwilde，当然，我正在阅读它talks.golang.org/2012/waza.slide#40 谢谢！
@tomwilde，在该示例中，Rob 没有缓冲频道，但如果我不使用作业总数缓冲频道，那么我只会得到工作完成后的工人数量。跨度>
啊，我忘记了一些基本的东西：如果没有某种缓冲，你必须在工人插入结果时耗尽结果。 play.golang.org/p/251IzyIMnu 启动一个耗尽的 goroutine。我的排水器只是将结果推送到列表中；在具有大量工作的真实工作池场景中，您可能希望排水器做一些有趣的事情（可以说是应用程序管道的下一个阶段）。最后，您也可以将作业插入器放在后台 goroutine 中，并将排水保持在前台。
避免无缓冲挂起的最简单方法：play.golang.org/p/7EdxQNlX33。只需在单独的 goroutine 中插入作业。 WaitGroup 确实是一个不相关的东西，我在同一个示例中添加两者有点混乱（也令人困惑）。

标签： mysql go

【解决方案1】：

您所拥有的基本工作，但要摆脱缓冲，您需要同时写入jobs 和读取results。否则，您的流程最终会卡住 - 工作人员无法发送结果，因为没有收到任何结果，并且您无法插入作业，因为工作人员被阻止。

这里是a boiled-down example on the Playground，介绍了如何在main 中接收结果时在后台推送作业的工作队列：

package main

import "fmt"

func worker(jobs <-chan int, results chan<- int) {
    for _ = range jobs {
        // ...do work here...
        results <- 1
    }
}

func main() {
    workers := 10
    total := 30
    jobs := make(chan int)
    results := make(chan int)

    // start workers
    for w := 0; w < workers; w++ {
        go worker(jobs, results)
    }

    // insert jobs in background
    go func() {
        for j := 0; j < total; j++ {
            jobs <- j
        }
    }()

    // collect results
    for i := 0; i < total; i++ {
        <-results
        fmt.Printf(".")
    }

    close(jobs)
}

要使特定代码正常工作，您必须知道会得到多少结果。如果您不知道（例如，每个作业可能产生零个或多个结果），您可以使用 sync.WaitGroup 到 wait for the workers to finish, then close the result stream：

package main

import (
    "fmt"
    "sync"
)

func worker(jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {
    for _ = range jobs {
        // ...do work here...
        results <- 1
    }
    wg.Done()
}

func main() {
    workers := 10
    total := 30
    jobs := make(chan int)
    results := make(chan int)
    wg := &sync.WaitGroup{}

    // start workers
    for w := 0; w < workers; w++ {
        wg.Add(1)
        go worker(jobs, results, wg)
    }

    // insert jobs in background
    go func() {
        for j := 0; j < total; j++ {
            jobs <- j
        }
        close(jobs)
        wg.Wait()
        // all workers are done so no more results
        close(results)
    }()

    // collect results
    for _ = range results {
        fmt.Printf(".")
    }

}

还有许多其他更复杂的技巧可以在发生错误后停止所有工作人员，将结果按与原始作业相同的顺序排列，或执行其他类似操作。不过，听起来好像基本版本在这里工作。

【讨论】：