我正在 Go 中进行一些流处理，并试图弄清楚如何在没有锁的情况下以“Go 方式”执行此操作。
这个人为的例子显示了我面临的问题。
- 我们一次得到一个
thing。 - 有一个 goroutine 将它们缓冲到一个名为
things. - 当
things变满时len(things) == 100，它会以某种方式处理并重置 - 有
n许多并发 goroutines 需要things在它满之前访问 - 从其他 goroutine访问“不完整”
things是不可预测的。 - 既不需要
doSomethingWithPartial也不doSomethingWithComplete需要变异things
代码：
var m sync.Mutex
var count int64
things := make([]int64, 0, 100)
// slices of data are constantly being generated and used
go func() {
for {
m.Lock()
if len(things) == 100 {
// doSomethingWithComplete does not modify things
doSomethingWithComplete(things)
things = make([]int64, 0, 100)
}
things = append(things, count)
m.Unlock()
count++
}
}()
// doSomethingWithPartial needs to access the things before they're ready
for {
m.Lock()
// doSomethingWithPartial does not modify things
doSomethingWithPartial(things)
m.Unlock()
}
我知道切片是不可变的，这是否意味着我可以删除互斥锁并期望它仍然有效（我假设没有）。我如何重构它以使用通道而不是互斥锁。
编辑：这是我想出的不使用互斥锁的解决方案
package main
import (
"fmt"
"sync"
"time"
)
func Incrementor() chan int {
ch := make(chan int)
go func() {
count := 0
for {
ch <- count
count++
}
}()
return ch
}
type Foo struct {
things []int
requests chan chan []int
stream chan int
C chan []int
}
func NewFoo() *Foo {
foo := &Foo{
things: make([]int, 0, 100),
requests: make(chan chan []int),
stream: Incrementor(),
C: make(chan []int),
}
go foo.Launch()
return foo
}
func (f *Foo) Launch() {
for {
select {
case ch := <-f.requests:
ch <- f.things
case thing := <-f.stream:
if len(f.things) == 100 {
f.C <- f.things
f.things = make([]int, 0, 100)
}
f.things = append(f.things, thing)
}
}
}
func (f *Foo) Things() []int {
ch := make(chan []int)
f.requests <- ch
return <-ch
}
func main() {
foo := NewFoo()
var wg sync.WaitGroup
wg.Add(10)
for i := 0; i < 10; i++ {
go func(i int) {
time.Sleep(time.Millisecond * time.Duration(i) * 100)
things := foo.Things()
fmt.Println("got things:", len(things))
wg.Done()
}(i)
}
go func() {
for _ = range foo.C {
// do something with things
}
}()
wg.Wait()
}