Go 并发切片访问答案

【问题标题】：Go concurrent slice accessGo 并发切片访问
【发布时间】：2013-04-18 16:46:51
【问题描述】：

我正在 Go 中进行一些流处理，但在试图找出如何在没有锁的情况下以“Go 方式”执行此操作时遇到了困难。

这个人为的例子显示了我面临的问题。

我们一次只收到一个thing。
有一个 goroutine 将它们缓冲到一个名为 things 的切片中。
当things 变满len(things) == 100 时，会以某种方式对其进行处理并重置
有 n 数量的并发 goroutine 需要访问 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()
}

【问题讨论】：

切片不是不可变的。字符串是。
切片是可变的。字符串是不可变的。
@FUZxxl 切片指向的数组是可变的，但切片本身不是。（AFAIK）
@ilia choly 这有点真实。请注意，我可以执行 *(&someslice) = someslice[a:b] 来覆盖 someslice 所在的内存位置。此外，如果您有一个全局变量是一个切片，您当然可以分配给它并更改切片全局变量包含。
@iliacholy myslice[5] = new_value?

标签： concurrency go

【解决方案1】：

需要注意的是，“Go way”可能只是为此使用互斥锁。弄清楚如何使用通道来做这件事很有趣，但对于这个特定问题，互斥锁可能更简单、更容易推理。

【讨论】：