Mehrere Goroutinen hören auf einem Kanal

82

Ich habe mehrere Goroutinen, die versuchen, gleichzeitig auf demselben Kanal zu empfangen. Es scheint, als würde die letzte Goroutine, die auf dem Kanal zu empfangen beginnt, den Wert erhalten. Ist dies irgendwo in der Sprachspezifikation oder ist es undefiniertes Verhalten?

c := make(chan string)
for i := 0; i < 5; i++ {
    go func(i int) {
        <-c
        c <- fmt.Sprintf("goroutine %d", i)
    }(i)
}
c <- "hi"
fmt.Println(<-c)

Ausgabe:

goroutine 4

Beispiel auf dem Spielplatz

BEARBEITEN:

Mir ist gerade klar geworden, dass es komplizierter ist als ich dachte. Die Nachricht wird an alle Goroutinen weitergegeben.

c := make(chan string)
for i := 0; i < 5; i++ {
    go func(i int) {
        msg := <-c
        c <- fmt.Sprintf("%s, hi from %d", msg, i)
    }(i)
}
c <- "original"
fmt.Println(<-c)

Ausgabe:

original, hi from 0, hi from 1, hi from 2, hi from 3, hi from 4

Beispiel auf dem Spielplatz

Ilia Choly
quelle
6
Ich habe Ihr letztes Snippet ausprobiert und (zu meiner großen Erleichterung) nur ausgegeben original, hi from 4...
Chang Qian
1
@ChangQian Das Hinzufügen eines time.Sleep(time.Millisecond)zwischen dem Senden und Empfangen des Kanals bringt das alte Verhalten zurück.
Ilia Choly

Antworten:

75

Ja, es ist kompliziert, aber es gibt ein paar Faustregeln, nach denen sich die Dinge viel einfacher anfühlen sollten.

  • Verwenden Sie lieber formale Argumente für die Kanäle, die Sie an Go-Routinen übergeben, als auf Kanäle im globalen Bereich zuzugreifen. Auf diese Weise können Sie mehr Compiler überprüfen und auch die Modularität verbessern.
  • Vermeiden Sie sowohl das Lesen als auch das Schreiben auf demselben Kanal in einer bestimmten Go-Routine (einschließlich der ' Haupt'-Routine ). Andernfalls ist Deadlock ein viel größeres Risiko.

Hier ist eine alternative Version Ihres Programms, die diese beiden Richtlinien anwendet. Dieser Fall zeigt viele Autoren und einen Leser auf einem Kanal:

c := make(chan string)

for i := 1; i <= 5; i++ {
    go func(i int, co chan<- string) {
        for j := 1; j <= 5; j++ {
            co <- fmt.Sprintf("hi from %d.%d", i, j)
        }
    }(i, c)
}

for i := 1; i <= 25; i++ {
    fmt.Println(<-c)
}

http://play.golang.org/p/quQn7xePLw

Es werden die fünf Go-Routinen erstellt, die in einen einzelnen Kanal schreiben, wobei jeder fünfmal schreibt. Die Hauptroutine liest alle fünfundzwanzig Nachrichten - Sie können feststellen, dass die Reihenfolge, in der sie angezeigt werden, häufig nicht sequentiell ist (dh die Parallelität ist offensichtlich).

Dieses Beispiel zeigt eine Funktion von Go-Kanälen: Es ist möglich, dass mehrere Autoren einen Kanal gemeinsam nutzen. Go verschachtelt die Nachrichten automatisch.

Gleiches gilt für einen Schreiber und mehrere Leser auf einem Kanal, wie im zweiten Beispiel hier gezeigt:

c := make(chan int)
var w sync.WaitGroup
w.Add(5)

for i := 1; i <= 5; i++ {
    go func(i int, ci <-chan int) {
        j := 1
        for v := range ci {
            time.Sleep(time.Millisecond)
            fmt.Printf("%d.%d got %d\n", i, j, v)
            j += 1
        }
        w.Done()
    }(i, c)
}

for i := 1; i <= 25; i++ {
    c <- i
}
close(c)
w.Wait()

Dieses zweite Beispiel enthält eine Wartezeit für die Hauptgoroutine, die andernfalls sofort beendet würde und dazu führen würde, dass die anderen fünf Goroutinen vorzeitig beendet werden (danke an olov für diese Korrektur) .

In beiden Beispielen war keine Pufferung erforderlich. Es ist im Allgemeinen ein gutes Prinzip, die Pufferung nur als Leistungsverbesserer anzusehen. Wenn Ihr Programm ohne Puffer nicht blockiert, blockiert es auch nicht mit Puffern (aber das Gegenteil ist nicht immer der Fall). So, wie eine andere Faustregel starten , ohne Pufferung dann fügen Sie später nach Bedarf .

Rick-777
quelle
Müssen Sie nicht warten, bis alle Goroutinen fertig sind?
mlbright
Es kommt darauf an, was du meinst. Schauen Sie sich die play.golang.org-Beispiele an. Sie haben eine mainFunktion, die beendet wird, sobald sie das Ende erreicht hat, unabhängig davon, was andere Goroutinen tun. Im ersten Beispiel oben mainist Lock-Step mit den anderen Goroutinen, so dass es kein Problem gibt. Das zweite Beispiel funktioniert ebenfalls problemlos, da alle Nachrichten über gesendet werden, c bevor die closeFunktion aufgerufen wird, und dies geschieht, bevor die mainGoroutine beendet wird. (Sie könnten argumentieren, dass das Anrufen closein diesem Fall überflüssig ist, aber es ist eine gute Praxis.)
Rick-777
1
Angenommen, Sie möchten (deterministisch) 15 Ausdrucke im letzten Beispiel sehen, müssen Sie warten. Um dies zu demonstrieren, ist hier das gleiche Beispiel, aber mit einer Zeit. Schlafen Sie kurz vor dem Printf
olov
Und hier ist das gleiche Beispiel mit einer Zeit. Schlafen und behoben mit einer WaitGroup, um auf die Goroutinen zu warten: play.golang.org/p/ESq9he_WzS
olov
Ich denke nicht, dass dies eine gute Empfehlung ist, die Pufferung zunächst wegzulassen. Ohne Pufferung schreiben Sie tatsächlich keinen gleichzeitigen Code, und dies führt nicht nur dazu, dass Sie keinen Deadlock durchführen können, sondern auch dazu, dass das Behandlungsergebnis von der anderen Seite des Kanals bereits bei der nächsten Anweisung nach dem Senden verfügbar ist, und Sie können sich unbeabsichtigt (oder absichtlich im Falle eines Neulings) darauf verlassen. Und wenn Sie sich darauf verlassen, dass Sie sofort ein Ergebnis haben, ohne speziell darauf zu warten, und einen Puffer hinzufügen, haben Sie eine Rennbedingung.
Benutzer
24

Späte Antwort, aber ich hoffe, dies hilft anderen in Zukunft wie Long Polling, "Global" Button, Broadcast an alle?

Effective Go erklärt das Problem:

Empfänger blockieren immer, bis Daten zu empfangen sind.

Das bedeutet, dass Sie nicht mehr als 1 Goroutine haben können, die 1 Kanal hört, und erwarten, dass ALLE Goroutinen den gleichen Wert erhalten.

Führen Sie dieses Codebeispiel aus .

package main

import "fmt"

func main() {
    c := make(chan int)

    for i := 1; i <= 5; i++ {
        go func(i int) {
        for v := range c {
                fmt.Printf("count %d from goroutine #%d\n", v, i)
            }
        }(i)
    }

    for i := 1; i <= 25; i++ {
        c<-i
    }

    close(c)
}

Sie werden "count 1" nicht mehr als einmal sehen, obwohl 5 Goroutinen den Kanal hören. Dies liegt daran, dass alle anderen Goroutinen in der Schlange warten müssen, wenn die erste Goroutine den Kanal blockiert. Wenn der Kanal entsperrt ist, wurde die Zählung bereits empfangen und aus dem Kanal entfernt, sodass die nächste Goroutine in der Zeile den nächsten Zählwert erhält.

Brenden
quelle
1
Danke - jetzt macht dieses Beispiel Sinn github.com/goinaction/code/blob/master/chapter6/listing20/…
user31208
Ahh das war hilfreich. Wäre es eine gute Alternative, für jede Go-Routine, die die Informationen benötigt, einen Kanal zu erstellen und bei Bedarf eine Nachricht auf allen Kanälen zu senden? Das ist die Option, die ich mir vorstellen kann.
ThePartyTurtle
8

Es ist kompliziert.

Sehen Sie auch, was mit passiert GOMAXPROCS = NumCPU+1. Zum Beispiel,

package main

import (
    "fmt"
    "runtime"
)

func main() {
    runtime.GOMAXPROCS(runtime.NumCPU() + 1)
    fmt.Print(runtime.GOMAXPROCS(0))
    c := make(chan string)
    for i := 0; i < 5; i++ {
        go func(i int) {
            msg := <-c
            c <- fmt.Sprintf("%s, hi from %d", msg, i)
        }(i)
    }
    c <- ", original"
    fmt.Println(<-c)
}

Ausgabe:

5, original, hi from 4

Und sehen Sie, was mit gepufferten Kanälen passiert. Zum Beispiel,

package main

import "fmt"

func main() {
    c := make(chan string, 5+1)
    for i := 0; i < 5; i++ {
        go func(i int) {
            msg := <-c
            c <- fmt.Sprintf("%s, hi from %d", msg, i)
        }(i)
    }
    c <- "original"
    fmt.Println(<-c)
}

Ausgabe:

original

Sie sollten auch diese Fälle erklären können.

peterSO
quelle
7

Ich habe vorhandene Lösungen untersucht und eine einfache Broadcast-Bibliothek https://github.com/grafov/bcast erstellt .

    group := bcast.NewGroup() // you created the broadcast group
    go bcast.Broadcasting(0) // the group accepts messages and broadcast it to all members

    member := group.Join() // then you join member(s) from other goroutine(s)
    member.Send("test message") // or send messages of any type to the group 

    member1 := group.Join() // then you join member(s) from other goroutine(s)
    val := member1.Recv() // and for example listen for messages
Alexander I.Grafov
quelle
2
Tolle Bibliothek, die du da hast! Ich habe auch github.com/asaskevich/EventBus
Benutzer
Und keine große Sache, aber vielleicht sollten Sie in der Readme-Datei erwähnen, wie Sie sich trennen können.
Benutzer
Speicherleck dort
jhvaras
:( Können Sie Details @jhvaras erklären?
Alexander I.Grafov
2

Bei mehreren Goroutinen auf einem Kanal hören, ja, das ist möglich. Der entscheidende Punkt ist die Nachricht selbst. Sie können eine Nachricht wie folgt definieren:

package main

import (
    "fmt"
    "sync"
)

type obj struct {
    msg string
    receiver int
}

func main() {
    ch := make(chan *obj) // both block or non-block are ok
    var wg sync.WaitGroup
    receiver := 25 // specify receiver count

    sender := func() {
        o := &obj {
            msg: "hello everyone!",
            receiver: receiver,
        }
        ch <- o
    }
    recv := func(idx int) {
        defer wg.Done()
        o := <-ch
        fmt.Printf("%d received at %d\n", idx, o.receiver)
        o.receiver--
        if o.receiver > 0 {
            ch <- o // forward to others
        } else {
            fmt.Printf("last receiver: %d\n", idx)
        }
    }

    go sender()
    for i:=0; i<reciever; i++ {
        wg.Add(1)
        go recv(i)
    }

    wg.Wait()
}

Die Ausgabe ist zufällig:

5 received at 25
24 received at 24
6 received at 23
7 received at 22
8 received at 21
9 received at 20
10 received at 19
11 received at 18
12 received at 17
13 received at 16
14 received at 15
15 received at 14
16 received at 13
17 received at 12
18 received at 11
19 received at 10
20 received at 9
21 received at 8
22 received at 7
23 received at 6
2 received at 5
0 received at 4
1 received at 3
3 received at 2
4 received at 1
last receiver 4
coanor
quelle