Plusieurs goroutines lisant à partir du même canal

l'éditeur php Strawberry vous présentera le contenu pertinent de plusieurs goroutines lisant depuis le même canal dans cet article. En programmation simultanée, goroutine est un thread léger du langage Go qui peut effectuer plusieurs tâches en même temps. Les canaux sont un moyen important de communiquer entre les goroutines. Lorsque plusieurs goroutines doivent lire des données du même canal, nous devons prêter attention à certains problèmes et prendre les mesures correspondantes pour garantir l'exactitude et l'efficacité du programme. Dans ce qui suit, nous expliquerons le processus en détail et fournirons quelques astuces et conseils pratiques.

Contenu de la question

Envisagez de créer plusieurs goroutines pour lire les valeurs du même canal. Les deux travailleurs sont générés comme prévu, mais ne lisent qu'un seul élément du canal et arrêtent la lecture. Je m'attends à ce que la goroutine continue de lire les données du canal jusqu'à ce que la goroutine envoyant la valeur au canal soit fermée. Bien que quelque chose empêche l'expéditeur d'envoyer, la goroutine qui a engendré le projet n'est pas fermée. Pourquoi chaque travailleur ne lit-il qu'une seule valeur et s'arrête-t-il ?

La sortie montre les deux valeurs envoyées, une lue par chaque goroutine de travail. La troisième valeur est envoyée mais n'est lue à partir d'aucun des threads de travail.

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

Aller au terrain de jeux

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:
        }
    }

}

Solution de contournement

Le commentaire précédent sur le fait que le canal n'est pas mis en mémoire tampon est correct, mais il existe d'autres problèmes de synchronisation.

Un canal sans tampon signifie essentiellement que lorsqu'une valeur est écrite, cette valeur doit être reçue avant que toute autre écriture puisse avoir lieu.

1. workerpool 创建一个无缓冲通道 out 来存储结果，但只有在所有结果写入 out 后才返回。但由于从 out 通道的读取发生在 out 返回之后，并且 out 没有缓冲，因此 workerpool 在尝试写入时被阻塞，从而导致死锁。这就是为什么看起来每个工作人员只发送一个值；实际上，在发送第一个之后，所有工作人员都被阻止，因为没有任何东西可以接收该值（您可以通过在写入 out Déplacez la déclaration d'impression après pour voir ceci)

Les options de correction incluent l'obligation pour out 有一个大小为 n = 结果数 的缓冲区（即 out := make(chan int, n)）或使 out 不缓冲并在写入时从 out de faire la lecture.

done 频道也没有被正确使用。 main 和 workerpool tous deux comptent sur lui pour arrêter l'exécution, mais rien n'y est écrit ! Il est également sans tampon et souffre donc du problème de blocage mentionné ci-dessus.

Pour résoudre ce problème, vous pouvez d'abord résoudre le blocage en passant workerpool 中删除 case <-done: 并简单地通过 in 进行范围，因为它在 main 中关闭。然后可以将doneà un canal mis en mémoire tampon.

Combinez ces correctifs pour obtenir :

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
        }
    }

}

Cela peut résoudre votre problème, mais ce n'est pas la meilleure façon d'utiliser la chaîne ! La structure elle-même peut être modifiée plus simplement sans avoir recours à des canaux tamponnés.

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!