Goroutines和相关函数

Question

there I am having some fun with GO and am just very curious about something I am trying to achieve. I have a package here that just gets a feed from Reddit noting special. When I receive the parent JSON file I would then like to retrieve child data. If you see the code below I launch a series of goroutines which I then block, waiting for them to finish using the sync package. What I would like is once the first series of goroutines finish the second series of goroutines using the previous results. There are a few was I was thinking such as for loop and switch statement. But what is the best and most efficient way to do this

func (m redditMatcher) retrieve(dataPoint *collect.DataPoint) (*redditCommentsDocument, error) {
    if dataPoint.URI == "" {
        return nil, errors.New("No datapoint uri provided")
    }

    // Get options data -> returns empty struct
    // if no options are present
    options := m.options(dataPoint.Options)
    if len(options.subreddit) <= 0 {
        return nil, fmt.Errorf("Matcher fail: Reddit - Subreddit option manditory
")
    }

    // Create an buffered channel to receive match results to display.
    results := make(chan *redditCommentsDocument, len(options.subreddit))

    // Generte requests for each subreddit produced using
    // goroutines concurency model
    for _, s := range options.subreddit {
        // Set the number of goroutines we need to wait for while
        // they process the individual subreddit.
        waitGroup.Add(1)
        go retrieveComment(s.(string), dataPoint.URI, results)
    }

    // Launch a goroutine to monitor when all the work is done.
    waitGroup.Wait()

    // HERE I WOULD TO CALL ANOTHER SERIES OFF GOROUTINES
    for commentFeed := range results {
        // HERE I WOULD LIKE TO CALL GO ROUTINES USING THE RESULTS
        // PROVIDED FROM THE PREVIOUS FUNCTIONS
        waitGroup.Add(1)
        log.Printf("%s

", commentFeed.Kind)
    }

    waitGroup.Wait()
    close(results)

    return nil, nil
}

Answer 1

If you want to wait for all of the first series to complete, then you can just pass in a pointer to your waitgroup, wait after calling all the first series functions (which will call Done() on the waitgroup), and then start the second series. Here's a runnable annotated code example that does that:

package main

import(
    "fmt"
    "sync"
    "time"
)

func first(wg *sync.WaitGroup) {
    defer wg.Done()
    fmt.Println("Starting a first")
    // do some stuff... here's a sleep to make some time pass
    time.Sleep(250 * time.Millisecond)
    fmt.Println("Done with a first")
}

func second(wg *sync.WaitGroup) {
    defer wg.Done()
    fmt.Println("Starting a second")
    // do some followup stuff
    time.Sleep(50 * time.Millisecond)
    fmt.Println("Done with a second")
}

func main() {
    wg := new(sync.WaitGroup) // you'll need a pointer to avoid a copy when passing as parameter to goroutine function

    // let's start 5 firsts and then wait for them to finish
    wg.Add(5)
    go first(wg)
    go first(wg)
    go first(wg)
    go first(wg)
    go first(wg)
    wg.Wait()

    // now that we're done with all the firsts, let's do the seconds
    // how about two of these
    wg.Add(2)
    go second(wg)
    go second(wg)
    wg.Wait()

    fmt.Println("All done")
}

It outputs:

Starting a first
Starting a first
Starting a first
Starting a first
Starting a first
Done with a first
Done with a first
Done with a first
Done with a first
Done with a first
Starting a second
Starting a second
Done with a second
Done with a second
All done

But if you want a "second" to start as soon as a "first" has finished, just have the seconds executing blocking receive operators on the channel while the firsts are running:

package main

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

func first(res chan int, wg *sync.WaitGroup) {
    defer wg.Done()
    fmt.Println("Starting a first")
    // do some stuff... here's a sleep to make some time pass
    time.Sleep(250 * time.Millisecond)
    fmt.Println("Done with a first")
    res <- rand.Int() // this will block until a second is ready
}

func second(res chan int, wg *sync.WaitGroup) {
    defer wg.Done()
    fmt.Println("Wait for a value from first")
    val := <-res // this will block until a first is ready
    fmt.Printf("Starting a second with val %d
", val)
    // do some followup stuff
    time.Sleep(50 * time.Millisecond)
    fmt.Println("Done with a second")
}

func main() {
    wg := new(sync.WaitGroup) // you'll need a pointer to avoid a copy when passing as parameter to goroutine function
    ch := make(chan int)

    // lets run first twice, and second once for each first result, for a total of four workers:
    wg.Add(4)
    go first(ch, wg)
    go first(ch, wg)
    // don't wait before starting the seconds
    go second(ch, wg)
    go second(ch, wg)
    wg.Wait()

    fmt.Println("All done")
}

Which outputs:

Wait for a value from first
Starting a first
Starting a first
Wait for a value from first
Done with a first
Starting a second with val 5577006791947779410
Done with a first
Starting a second with val 8674665223082153551
Done with a second
Done with a second
All done