Mehrere Goroutinen lesen vom selben Kanal

php-Editor Strawberry stellt Ihnen in diesem Artikel den relevanten Inhalt mehrerer Goroutinen vor, die aus demselben Kanal lesen. Bei der gleichzeitigen Programmierung ist Goroutine ein leichter Thread in der Go-Sprache, der mehrere Aufgaben gleichzeitig ausführen kann. Kanäle sind eine wichtige Möglichkeit zur Kommunikation zwischen Goroutinen. Wenn mehrere Goroutinen Daten aus demselben Kanal lesen müssen, müssen wir auf einige Probleme achten und entsprechende Maßnahmen ergreifen, um die Korrektheit und Effizienz des Programms sicherzustellen. Im Folgenden erläutern wir den Vorgang im Detail und geben einige praktische Tipps und Ratschläge.

Frageninhalt

Erwägen Sie, mehrere Goroutinen zu erzeugen, um Werte aus demselben Kanal zu lesen. Die beiden Worker werden wie erwartet generiert, lesen jedoch nur ein Element aus dem Kanal und hören auf zu lesen. Ich gehe davon aus, dass die Goroutine weiterhin Daten vom Kanal liest, bis die Goroutine, die den Wert an den Kanal sendet, geschlossen wird. Obwohl der Absender durch etwas am Senden gehindert wird, ist die Goroutine, die das Projekt hervorgebracht hat, nicht geschlossen. Warum liest jeder Arbeiter nur einen Wert und stoppt?

Die Ausgabe zeigt die beiden gesendeten Werte, einen, der von jeder Worker-Goroutine gelesen wird. Der dritte Wert wird gesendet, aber von keinem der Arbeitsthreads gelesen.

new worker
new worker
waiting
sending 0
sending 1
sending 2
running func 1
sending value out 1
running func 0
sending value out 0

Gehen Sie auf den Spielplatz

package main

import (
    "fmt"
    "sync"
)

func workerPool(done <-chan bool, in <-chan int, numberOfWorkers int, fn func(int) int) chan int {
    out := make(chan int)
    var wg sync.WaitGroup

    for i := 0; i < numberOfWorkers; i++ {
        fmt.Println("new worker")
        wg.Add(1)
        // fan out worker goroutines reading from in channel and
        // send output into out channel
        go func() {
            defer wg.Done()
            for {
                select {
                case <-done:
                    fmt.Println("recieved done signal")
                    return
                case data, ok := <-in:
                    if !ok {
                        fmt.Println("no more items")
                        return
                    }
                    // fan-in job execution multiplexing results into the results channel
                    fmt.Println("running func", data)
                    value := fn(data)
                    fmt.Println("sending value out", value)
                    out <- value
                }
            }
        }()
    }

    fmt.Println("waiting")
    wg.Wait()
    fmt.Println("done waiting")
    close(out)
    return out
}

func main() {
    done := make(chan bool)
    defer close(done)

    in := make(chan int)

    go func() {
        for i := 0; i < 10; i++ {
            fmt.Println("sending", i)
            in <- i
        }
        close(in)
    }()

    out := workerPool(done, in, 2, func(i int) int {
        return i
    })

    for {
        select {
        case o, ok := <-out:
            if !ok {
                continue
            }

            fmt.Println("output", o)
        case <-done:
            return
        default:
        }
    }

}

Workaround

Der vorherige Kommentar, dass der Kanal nicht gepuffert wird, ist richtig, aber es gibt andere Synchronisierungsprobleme.

Ein ungepufferter Kanal bedeutet im Wesentlichen, dass beim Schreiben eines Werts dieser Wert empfangen werden muss, bevor andere Schreibvorgänge erfolgen können.

1. workerpool 创建一个无缓冲通道 out 来存储结果，但只有在所有结果写入 out 后才返回。但由于从 out 通道的读取发生在 out 返回之后，并且 out 没有缓冲，因此 workerpool 在尝试写入时被阻塞，从而导致死锁。这就是为什么看起来每个工作人员只发送一个值；实际上，在发送第一个之后，所有工作人员都被阻止，因为没有任何东西可以接收该值（您可以通过在写入 out Verschieben Sie die Druckanweisung nach hinten, um dies zu sehen)

Zu den

Fix-Optionen gehört, out 有一个大小为 n = 结果数 的缓冲区（即 out := make(chan int, n)）或使 out 不缓冲并在写入时从 out das Lesen durchführen zu lassen.

1. done 频道也没有被正确使用。 main 和 workerpool beide verlassen sich darauf, um die Ausführung zu stoppen, aber es ist nichts darauf geschrieben! Es ist außerdem ungepuffert und weist daher das oben erwähnte Deadlock-Problem auf.

Um dieses Problem zu beheben, können Sie zunächst den Deadlock von der workerpool 中删除 case <-done: 并简单地通过 in 进行范围，因为它在 main 中关闭。然后可以将doneEinstellung auf einen gepufferten Kanal beheben.

Kombinieren Sie diese Korrekturen, um Folgendes zu erhalten:

package main

import (
    "fmt"
    "sync"
)

func workerPool(done chan bool, in <-chan int, numberOfWorkers int, fn func(int) int) chan int {
    out := make(chan int, 100)
    var wg sync.WaitGroup

    for i := 0; i < numberOfWorkers; i++ {
        fmt.Println("new worker")
        wg.Add(1)
        // fan out worker goroutines reading from in channel and
        // send output into out channel
        go func() {
            defer wg.Done()
            for data := range in {
                // fan-in job execution multiplexing results into the results channel
                fmt.Println("running func", data)
                value := fn(data)
                fmt.Println("sending value out", value)
                out <- value

            }
            fmt.Println("no more items")
            return
        }()
    }

    fmt.Println("waiting")
    wg.Wait()
    fmt.Println("done waiting")
    close(out)
    done <- true
    close(done)
    return out
}

func main() {
    done := make(chan bool, 1)

    in := make(chan int)

    go func() {
        for i := 0; i < 10; i++ {
            fmt.Println("sending", i)
            in <- i
        }
        close(in)
    }()

    out := workerPool(done, in, 2, func(i int) int {
        return i
    })

    for {
        select {
        case o, ok := <-out:
            if !ok {
                continue
            }

            fmt.Println("output", o)
        case <-done:
            return
        }
    }

}

Das löst vielleicht Ihr Problem, ist aber nicht die beste Art, den Kanal zu nutzen! Die Struktur selbst kann einfacher geändert werden, ohne auf gepufferte Kanäle angewiesen zu sein.

Das obige ist der detaillierte Inhalt vonMehrere Goroutinen lesen vom selben Kanal. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!