我有 30'000 个字符串的切片。我如何将这个切片分成 10 个 goroutine，从切片中提取 3000 个字符串，从中提取一些数据并推送到一个新切片中？
所以，最后，我将有 10 个切片，每个切片有 3000 个处理结果。处理这个问题的模式是什么？
我查看了this article，但不确定这些模式中的哪一个适用于我的案例。
【问题讨论】：
使用通道，从切片中读取元素，使用扇出来分配负载并传递消息。然后，在 goroutine 中处理字符串并在单个 goroutine 中收集结果（扇入）以避免互斥。
您可能想要设置最大并发并发 goroutine 的数量。
请记住，切片在写入时不是线程安全的。
有用的信息：
https://blog.golang.org/pipelines https://talks.golang.org/2012/concurrency.slide#1 https://blog.golang.org/advanced-go-concurrency-patterns https://talks.golang.org/2013/advconc.slide#1
【讨论】：
我同意@JimB 的观点，即为什么要限制 go 例程。但是，既然这是您的询问，我可能会这样。 [[3000 items],[3000 items],..] 然后在该二维数组中的每个索引都有 1 个常规过程。否则，下面只会将 goroutines 限制为 10。 方法一 主包
import (
"crypto/rand"
"fmt"
"log"
"sync"
"time"
)
var s []string
// genetate some mock data
func init() {
s = make([]string, 30000)
n := 5
for i := 0; i < 30000; i++ {
b := make([]byte, n)
if _, err := rand.Read(b); err != nil {
panic(err)
}
s[i] = fmt.Sprintf("%X", b)
}
}
func main() {
// set the number of workers
ch := make(chan string)
var mut sync.Mutex
counter := 0
// limit the number of gorountines to 10
for w := 0; w < 10; w++ {
go func(ch chan string, mut *sync.Mutex) {
for {
// get and update counter using mux to stop race condtions
mut.Lock()
i := counter
counter++
mut.Unlock()
// break the loop
if counter > len(s) {
return
}
// get string
myString := s[i]
// to some work then pass to channel
ch <- myString
}
}(ch, &mut)
}
// adding time. If you play wiht the number of gorountines youll see how changing the number above can efficiency
t := time.Now()
for i := 0; i < len(s); i++ {
result := <-ch
log.Println(time.Since(t), result, i)
}
}
METHOD2 init 函数正在创建一个 2d 数组，该数组被分成 10 个数组，每个数组包含 3000 个元素。如果您以这种方式解析数据，那么下面的逻辑只需很少修改即可工作
package main
import (
"crypto/rand"
"fmt"
"log"
"sync"
)
var chunkedSlice [10][3000]string
// genetate some mock data
// 2d array, each chunk has 3000 items in it
// there are 10 chunks, 1 go rountine per chunk
func init() {
n := 5
for i := 0; i < 10; i++ {
for j := 0; j < 3000; j++ {
b := make([]byte, n)
if _, err := rand.Read(b); err != nil {
panic(err)
}
chunkedSlice[i][j] = fmt.Sprintf("%X", b)
}
}
}
func main() {
// channel to send parsed data to
ch := make(chan string)
var wg sync.WaitGroup
// 10 chunks
for _, chunk := range chunkedSlice {
wg.Add(1)
// if defining the 2d array e.g [10][3000]string, you need to pass it as a pointer to avoid stack error
go func(ch chan string, wg *sync.WaitGroup, chunk *[3000]string) {
defer wg.Done()
for i := 0; i < len(chunk); i++ {
str := chunk[i]
// fmt.Println(str)
// parse the data (emulating)
parsedData := str
// send parsed data to the channel
ch <- parsedData
}
}(ch, &wg, &chunk)
}
// wait for all the routines to finish and close the channel
go func() {
wg.Wait()
close(ch)
}()
var counter int // adding to check that the right number of items was parsed
// get the data from the channel
for res := range ch {
log.Println(res, counter)
counter++
}
}
【讨论】：
w 子切片并处理每个子切片会更容易，而不是使用 goroutine。或者在扇入输出通道之外使用扇出输入通道。
在不同的 goroutine 中处理不同的内存地址是没有问题的。如果您的读写操作不会掩盖块，那么您可以简单地使用地块中的范围。 https://play.golang.org/p/MU5njoNmIf7
【讨论】：
我开发了一个库parapipe 来解决这样的任务。只需将切片分成 10 个部分并将它们发送到 Pipeline，它会同时处理它们：
import "github.com/nazar256/parapipe"
//...
var longStringSlice []string
// ...
pipeline := parapipe.NewPipeline(10).
Pipe(func(msg interface{}) interface{} {
slicePart := msg.([]string)
// process here to result
return result
})
// chop the slice and stream parts
chopSize := int(math.Ceil(float64(len(longStringSlice)) / 10))
for i:=0;i<10;i++ {
firstIdx := i * chopSize
lastIdx := (i+1) * chopSize
if lastIdx > len(longStringSlice) {
lastIdx = len(longStringSlice)
}
pipeline.In() <- longStringSlice[firstIdx:lastIdx]
}
【讨论】：