Golang Nebenläufigkeit: wie an die gleiche Scheibe aus verschiedenen goroutines anhängen

Question

Ich habe gleichzeitige goroutines, die eine (Zeiger auf a) struct an die gleiche Scheibe anhängen möchten. Wie schreibt man das in Go, um es nebenläufigkeitssicher zu machen?

Dies würde meine Concurrency-unsicheren Code sein, eine Wartegruppe mit:

var wg sync.WaitGroup 
MySlice = make([]*MyStruct) 
for _, param := range params { 
    wg.Add(1) 
    go func(param string) { 
     defer wg.Done() 
     OneOfMyStructs := getMyStruct(param) 
     MySlice = append(MySlice, &OneOfMyStructs) 
    }(param) 
} 
wg.Wait() 

Ich denke, man würde Kanäle für Concurrency-Sicherheit verwenden müssen. Kann jemand mit einem Beispiel dazu beitragen?

Answer 1

Ein Kanal ist der beste Weg, um dies anzugehen. Hier ist ein Beispiel, das unter go playground ausgeführt werden kann.

package main 

import "fmt" 
import "sync" 
import "runtime" 

type T int 

func main() { 
    var slice []T 
    var wg sync.WaitGroup 

    queue := make(chan T, 1) 

    // Create our data and send it into the queue. 
    wg.Add(100) 
    for i := 0; i < 100; i++ { 
     go func(i int) { 
      defer wg.Done() 

      // Do stuff. 
      runtime.Gosched() 

      queue <- T(i) 
     }(i) 
    } 

    // Poll the queue for data and append it to the slice. 
    // Since this happens synchronously and in the same 
    // goroutine/thread, this can be considered safe. 
    go func() { 
     defer wg.Done() 
     for t := range queue { 
      slice = append(slice, t) 
     } 
    }() 

    // Wait for everything to finish. 
    wg.Wait() 

    fmt.Println(slice) 
} 

Hinweis: Der runtime.Gosched() Aufruf ist es, weil diese goroutines an den Scheduler nicht nachgeben. Das würde einen Deadlock verursachen, wenn wir nicht explizit etwas tun, um den Scheduler auszulösen. Eine andere Option könnte gewesen sein, einige I/O durchzuführen (z. B. print to stdout). Aber ich finde eine runtime.Gosched() einfacher und klarer in ihrer Absicht.

Answer 2

Es ist nichts falsch mit der Bewachung der MySlice = append(MySlice, &OneOfMyStructs) mit einem sync.Mutex. Aber natürlich können Sie einen Ergebnis-Kanal mit der Puffergröße len(params) haben, alle goroutines senden ihre Antworten und sobald Ihre Arbeit beendet ist, sammeln Sie von diesem Ergebniskanal.

Wenn Ihr params eine feste Größe hat:

MySlice = make([]*MyStruct, len(params)) 
for i, param := range params { 
    wg.Add(1) 
    go func(i int, param string) { 
     defer wg.Done() 
     OneOfMyStructs := getMyStruct(param) 
     MySlice[i] = &OneOfMyStructs 
    }(i, param) 
} 

Da alle goroutines auf verschiedene Speicher schreiben ist dies nicht rassig.

Answer 3

Die Antwort, die von @jimt gepostet wird, ist nicht ganz richtig, da es den letzten im Kanal gesendeten Wert verfehlt und der letzte defer wg.Done() nie aufgerufen wird. Der folgende Ausschnitt enthält die Korrekturen.

https://play.golang.org/p/7N4sxD-Bai

package main 

import "fmt" 
import "sync" 

type T int 

func main() { 
    var slice []T 
    var wg sync.WaitGroup 

    queue := make(chan T, 1) 

    // Create our data and send it into the queue. 
    wg.Add(100) 
    for i := 0; i < 100; i++ { 
     go func(i int) { 
      // defer wg.Done() <- will result in the last int to be missed in the receiving channel 
      queue <- T(i) 
     }(i) 
    } 

    go func() { 
     // defer wg.Done() <- Never gets called since the 100 `Done()` calls are made above, resulting in the `Wait()` to continue on before this is executed 
     for t := range queue { 
      slice = append(slice, t) 
      wg.Done() // ** move the `Done()` call here 
     } 
    }() 

    wg.Wait() 

    // now prints off all 100 int values 
    fmt.Println(slice) 
} 

Answer 4

Sie entweder Daniele-Lösung und ein WaitGroup, das heißt verwenden, wenn die endgültige Scheibengröße bekannt ist, oder Sie use a channel zum dynamischen Elemente asynchron und dann wachsen die Scheibe zu sammeln.

Sie brauchen dann keine zusätzliche WaitGroup. Kanäle bieten bereits die erforderlichen Synchronisierungsfunktionen. Sie müssen den Kanal nur schließen, nachdem Sie Ihre Daten gelesen haben.

package main 

import "fmt" 

type T int 

func main() { 
    var slice []T 
    queue := make(chan T, 1) 

    // concurrently produce some data 
    for i := 0; i < 20; i++ { 
     go func(i int) { 
      queue <- T(i) 
     }(i) 
    } 

    remaining := 20 
    for t := range queue { 
     // This loop blocks until a new item is available in the channel. 
     // You can grow your slice here, but must also care to close the 
     // channel, when you decide that you obtained enough data. 
     if t != 13 { // because it is evil ;) 
      slice = append(slice, t) 
     } 
     if remaining--; remaining == 0 { 
      close(queue) // do not forget to close the channel 
     } 
    } 

    fmt.Printf("slice: %v, len: %v", slice, len(slice)) 
}