Event-driven-Muster in golang

BEARBEITEN: vielleicht habe ich deine Frage falsch verstanden, ich sehe, dass Sie erwähnen, dass "main" beginnen viele Produzent goroutines. Ich dachte, es war viele Verbraucher goroutines und einem einzigen Hersteller. Verlassen Sie die Antwort hier im Fall kann es sinnvoll sein, für andere auf, die für das Muster, aber die Kugel Punkte noch für Ihren Fall.

Also, wenn ich das richtig verstehe ist dein use-case ist, können Sie nicht erwarten, um zu senden auf einem Kanal, und Lesen Sie die Ergebnisse sofort nach. Sie wissen nicht, Wann der worker-Prozess senden, müssen Sie die Kommunikation zwischen goroutines, und dass geschieht mit den Kanälen. Angenommen, nur den Aufruf einer Funktion mit Rückgabewert funktioniert nicht in Ihrem Szenario, wenn Sie wirklich brauchen, um zu senden zu einem Arbeiter, dann blockieren Sie bis Sie das Ergebnis hat, könnten Sie schicken ein Kanal als Teil der Datenstruktur, und block-erhalten Sie nach dem senden, also:

resCh := make(chan Result)
ch <- Data{key, value, resCh}
res := <- resCh

Aber Sie sollten wahrscheinlich versuchen zu brechen, der das Werk als eine Reihe von voneinander unabhängigen Schritten statt, siehe den blog-post, den ich verlinkt in der ursprünglichen Antwort.

Ursprüngliche Antwort, wo ich dachte, es war eine einzelne Produzent - mehrere Verbraucher/Arbeitnehmer Muster:

Dies ist ein gemeinsames Muster für die Go goroutines und Kanälen Semantik sehr gut geeignet sind. Es gibt ein paar Dinge, die Sie benötigen, im Auge zu behalten:

• Der main-Funktion wird nicht automatisch warten goroutines zu beenden. Wenn es nichts anderes zu tun in der Haupt -, dann wird das Programm beendet und Sie nicht über Ihre Ergebnisse.

• Die Globale Karte, die Sie verwenden, ist nicht thread-safe. Müssen Sie synchronisieren, Zugang über ein mutex, aber es gibt einen besseren Weg - mit einem output-Kanal für die Ergebnisse, die bereits synchronisiert.

• Können Sie mithilfe einer for..über einen Kanal, und Sie können sicher teilen sich ein Kanal zwischen mehreren goroutines. Wie wir sehen werden, macht, dass dieses Muster sehr elegant, zu schreiben.

Spielplatz: https://play.golang.org/p/WqyZfwldqp

Mehr auf Gehen Rohrleitungen und Parallelität-Muster, Einführung, Fehlerbehandlung, die vorzeitige Kündigung, etc.: https://blog.golang.org/pipelines

Kommentierte code für die Einsatz-Fall, die Sie erwähnen:

//could be a command-line flag, a config, etc.
const numGoros = 10

//Data is a similar data structure to the one mentioned in the question.
type Data struct {
    key   string
    value int
}

func main() {
    var wg sync.WaitGroup

    //create the input channel that sends work to the goroutines
    inch := make(chan Data)
    //create the output channel that sends results back to the main function
    outch := make(chan Data)

    //the WaitGroup keeps track of pending goroutines, you can add numGoros
    //right away if you know how many will be started, otherwise do .Add(1)
    //each time before starting a worker goroutine.
    wg.Add(numGoros)
    for i := 0; i < numGoros; i++ {
        //because it uses a closure, it could've used inch and outch automaticaly,
        //but if the func gets bigger you may want to extract it to a named function,
        //and I wanted to show the directed channel types: within that function, you
        //can only receive from inch, and only send (and close) to outch.
        //
        //It also receives the index i, just for fun so it can set the goroutines'
        //index as key in the results, to show that it was processed by different
        //goroutines. Also, big gotcha: do not capture a for-loop iteration variable
        //in a closure, pass it as argument, otherwise it very likely won't do what
        //you expect.
        go func(i int, inch <-chan Data, outch chan<- Data) {
            //make sure WaitGroup.Done is called on exit, so Wait unblocks
            //eventually.
            defer wg.Done()

            //range over a channel gets the next value to process, safe to share
            //concurrently between all goroutines. It exits the for loop once
            //the channel is closed and drained, so wg.Done will be called once
            //ch is closed.
            for data := range inch {
                //process the data...
                time.Sleep(10 * time.Millisecond)
                outch <- Data{strconv.Itoa(i), data.value}
            }
        }(i, inch, outch)
    }

    //start the goroutine that prints the results, use a separate WaitGroup to track
    //it (could also have used a "done" channel but the for-loop would be more complex, with a select).
    var wgResults sync.WaitGroup
    wgResults.Add(1)
    go func(ch <-chan Data) {
        defer wgResults.Done()

        //to prove it processed everything, keep a counter and print it on exit
        var n int
        for data := range ch {
            fmt.Println(data.key, data.value)
            n++
        }

        //for fun, try commenting out the wgResults.Wait() call at the end, the output
        //will likely miss this line.
        fmt.Println(">>> Processed: ", n)
    }(outch)

    //send work, wherever that comes from...
    for i := 0; i < 1000; i++ {
        inch <- Data{"main", i}
    }

    //when there's no more work to send, close the inch, so the goroutines will begin
    //draining it and exit once all values have been processed.
    close(inch)

    //wait for all goroutines to exit
    wg.Wait()

    //at this point, no more results will be written to outch, close it to signal
    //to the results goroutine that it can terminate.
    close(outch)

    //and wait for the results goroutine to actually exit, otherwise the program would
    //possibly terminate without printing the last few values.
    wgResults.Wait()
}

In real-life-Szenarien, wo die Menge der Arbeit ist nicht bekannt, vor der Zeit, die Schließung der in-Kanal kann sich z.B. ein SIGINT-signal. Nur stellen Sie sicher, dass keine code-Pfad schicken kann funktionieren, nachdem der Kanal geschlossen war, denn das würde die Panik.