select - 选择中的 Golang channel 未接收

Question

我目前正在编写一个小脚本，我在其中使用 channel 、选择和 goroutine，我真的不明白为什么它没有像我想的那样运行。

我有 2 个 channel 供我所有的 goroutines 收听。

我将 channel 传递给每个 goroutine，其中有一个 select，必须根据数据首先出现的位置在 2 个之间进行选择。

问题是没有goroutine落入第二种情况。我可以一个接一个地收到 100 个工作，我在日志中看到了所有内容。它很好地完成了第一种情况下的要求，然后它在第二个 channel 中发送了工作(如果它做得很好......)我没有更多的日志。我只是不明白为什么...

如果有人能启发我:)

package main

func main() {

    wg := new(sync.WaitGroup)
    in := make(chan *Job)
    out := make(chan *Job)
    results := make(chan *Job)

    for i := 0; i < 50; i++ {
        go work(wg, in, out, results)
    }

    wg.Wait()

    // Finally we collect all the results of the work.
    for elem := range results {
            fmt.Println(elem)
    }    
}

func Work(wg *sync.WaitGroup, in chan *Job, out chan *Job, results chan *Job) {
    wg.Add(1)
    defer wg.Done()
    for {
        select {
        case job := <-in:
            ticker := time.Tick(10 * time.Second)

            select {
            case <-ticker:
                // DO stuff
            if condition is true {
                out <- job
            }
            case <-time.After(5 * time.Minute):
                fmt.Println("Timeout")
            }
        case job := <-out:
            ticker := time.Tick(1 * time.Minute)

            select {
            case <-ticker:
                // DO stuff
            if condition is true {
                results <- job
            }

            case <-quitOut:
                fmt.Println("Job completed")
            }
        }
    }
}

我创建了一些监听 2 个 channel 并将最终结果发送到第 3 个 channel 的 worker 。

它对收到的作业做一些事情，如果它验证了给定的条件，它会将此作业传递到下一个 channel ，如果它验证了一个条件，它会将作业传递到结果 channel 。

所以，在我的脑海中，我有一个这样的管道，例如 5 个工作人员: channel IN 中的 3 个工作，直接由 3 个工作人员接管它们，如果这 3 个工作验证条件，它们将在 channel OUT 中发送。直接由 2 名 worker 接手，第 3 份工作由前 3 名 worker 中的一名接手 ...

现在我希望你对我的第一段代码有更好的理解。但在我的代码中，我从来没有遇到过第二种情况。

Answer 1

我认为您的解决方案可能有点过于复杂。这是一个简化版本。请记住，有许多实现。一篇值得阅读的好文章

https://medium.com/smsjunk/handling-1-million-requests-per-minute-with-golang-f70ac505fcaa

或者直接从 Go 手册中更好

https://gobyexample.com/worker-pools (我认为这可能是您的目标)

无论如何，下面是一个不同类型的例子。有几种方法可以解决这个问题。

package main

import (
    "context"
    "log"
    "os"
    "sync"
    "time"
)

type worker struct {
    wg   *sync.WaitGroup
    in   chan job
    quit context.Context
}

type job struct {
    message int
}

func main() {
    numberOfJobs := 50

    ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
    defer cancel()

    w := worker{
        wg:   &sync.WaitGroup{},
        in:   make(chan job),
        quit: ctx,
    }

    for i := 0; i < numberOfJobs; i++ {
        go func(i int) {
            w.in <- job{message: i}
        }(i)
    }

    counter := 0
    for {
        select {
        case j := <-w.in:
            counter++
            log.Printf("Received job %+v\n", j)
            // DO SOMETHING WITH THE RECEIVED JOB
            // WORKING ON IT
            x := j.message * j.message
            log.Printf("job processed, result %d", x)
        case <-w.quit.Done():
            log.Printf("Recieved quit, timeout reached.  Number of jobs queued: %d, Number of jobs complete: %d\n", numberOfJobs, counter)
            os.Exit(0)
        default:
            // TODO
        }
    }

}