Đồng thời Golang: cách nối thêm vào cùng một lát từ các goroutines khác nhau

Question

Tôi có các goroutine đồng thời muốn nối thêm một cấu trúc (con trỏ đến a) vào cùng một lát. Làm thế nào để bạn viết rằng trong Go để làm cho nó đồng thời an toàn?

này sẽ được mã đồng thời-không an toàn của tôi, sử dụng một nhóm chờ đợi:

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() 

Tôi đoán bạn sẽ cần phải sử dụng đi các kênh truyền hình cho đồng thời an toàn. Bất cứ ai có thể đóng góp một ví dụ?

Answer 1

Kênh là cách tốt nhất để giải quyết vấn đề này. Dưới đây là một ví dụ có thể chạy trên go playground.

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

Note: Cuộc gọi runtime.Gosched() là có bởi vì những goroutines không chịu khuất phục trước Scheduler. Mà sẽ gây ra một bế tắc nếu chúng ta không rõ ràng làm một cái gì đó để kích hoạt lịch trình nói. Một tùy chọn khác có thể là để thực hiện một số I/O (ví dụ: in tới stdout). Nhưng tôi tìm thấy một runtime.Gosched() để dễ dàng và rõ ràng hơn trong mục đích của nó.

Answer 2

Không có gì sai khi bảo vệ MySlice = append(MySlice, &OneOfMyStructs) bằng sync.Mutex. Nhưng tất nhiên bạn có thể có một kênh kết quả với kích thước bộ đệm len(params) tất cả các goroutines gửi câu trả lời của họ và sau khi công việc của bạn kết thúc, bạn thu thập từ kênh kết quả này.

Nếu params của bạn có kích thước cố định:

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

Như tất cả goroutines ghi vào bộ nhớ khác nhau này là không hấp dẫn.

Answer 3

Câu trả lời được đăng bởi @jimt không hoàn toàn đúng, trong đó nó bỏ lỡ giá trị cuối cùng được gửi trong kênh và số defer wg.Done() cuối cùng không bao giờ được gọi. Đoạn mã dưới đây có các chỉnh sửa.

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

Bạn có thể sử dụng giải pháp Daniele và một WaitGroup, ví dụ, nếu kích thước lát cuối cùng được biết đến, hoặc bạn use a channel để thu thập các mặt hàng đồng bộ và sau đó phát triển slice động.

Bạn không cần thêm một nhóm WaitGroup nữa. Các kênh đã cung cấp khả năng đồng bộ hóa cần thiết. Bạn chỉ cần đóng kênh sau khi đọc dữ liệu của bạn.

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