I made some edits from the gobyexample:

import (
    "fmt"
    "math/rand"
    "time"
)

type DemoResult struct {
    Name string
    Rate int
}

func random(min, max int) int {
    rand.Seed(time.Now().UTC().UnixNano())
    return rand.Intn(max-min)   min
}

func worker(id int, jobs <-chan int, results chan<- DemoResult) {
    for j := range jobs {
        fmt.Println("worker", id, "started  job", j)
        time.Sleep(time.Second)
        fmt.Println("worker", id, "finished job", j)
        myrand := random(1, 4)
        if myrand == 2 {
            results <- DemoResult{Name: "succ", Rate: j}
        }
        //  else {
        //  results <- DemoResult{Name: "failed", Rate: 999}
        // }
    }
}

func main() {
    const numJobs = 5
    jobs := make(chan int, numJobs)
    results := make(chan DemoResult)
    for w := 1; w <= 3; w   {
        go worker(w, jobs, results)
    }
    for j := 1; j <= numJobs; j   {
        jobs <- j
    }
    close(jobs)

    for a := 1; a <= numJobs; a   {
        out := <-results
        if out.Name == "succ" {
            fmt.Printf("%v\n", out)
        }
    }
}

I commented the following code intentional to make it stuck forever:

        //  else {
        //  results <- DemoResult{Name: "failed", Rate: 999}
        // }

It seems like we should make the result's length the same as jobs'. I was wondering if we could make it have different length?

CodePudding user response：

Use a wait group to detect when the workers are done. Close the results channel when the workers are done. Receive results until the channel is closed.

func worker(wg *sync.WaitGroup, id int, 
            jobs <-chan int, 
            results chan<- DemoResult) {
    // Decrement wait group counter on return from
    // function.
    defer wg.Done()
    ⋮ 
}


func main() {
    ⋮
    // Declare wait group and increment counter for
    // each worker.
    var wg sync.WaitGroup
    for w := 1; w <= 3; w   {
        wg.Add(1)
        go worker(&wg, w, jobs, results)
    }
    ⋮
    // Wait for workers to decrement wait group
    // counter to zero and close channel.
    // Execute in goroutine so we can continue on 
    // to receiving values from results in main. 
    go func() {
        wg.Wait()
        close(results)
    }()
    ⋮
    // Loop until results is closed.
    for out := range results {
       ⋮
    }
}

https://go.dev/play/p/FOQwybMl7tM

CodePudding user response：

I was wondering if we could make it have different length?

Absolutely but you need some way of determining when you have reached the end of the results. This is the reason your example fails - currently the function assumes there will be numJobs (one result per job) results and waits for that many.

An alternative would be to use the channels closure to indicate this i.e. (playground)

package main

import (
    "fmt"
    "math/rand"
    "sync"
    "time"
)

type DemoResult struct {
    Name string
    Rate int
}

func random(min, max int) int {
    rand.Seed(time.Now().UTC().UnixNano())
    return rand.Intn(max-min)   min
}

func worker(id int, jobs <-chan int, results chan<- DemoResult) {
    for j := range jobs {
        fmt.Println("worker", id, "started  job", j)
        time.Sleep(time.Second)
        fmt.Println("worker", id, "finished job", j)
        myrand := random(1, 4)
        if myrand == 2 {
            results <- DemoResult{Name: "succ", Rate: j}
        } // else {
        //  results <- DemoResult{Name: "failed", Rate: 999}
        //}
    }
}

func main() {
    const numWorkers = 3
    const numJobs = 5
    jobs := make(chan int, numJobs)
    results := make(chan DemoResult)

    var wg sync.WaitGroup
    wg.Add(numWorkers)
    for w := 1; w <= numWorkers; w   {
        go func() {
            worker(w, jobs, results)
            wg.Done()
        }()
    }
    go func() {
        wg.Wait() // Wait for go routines to complete then close results channel
        close(results)
    }()

    for j := 1; j <= numJobs; j   {
        jobs <- j
    }
    close(jobs)

    for out := range results {
        if out.Name == "succ" {
            fmt.Printf("%v\n", out)
        }
    }
}