与多个生产者/多个消费者并发

Question

我可能遗漏了一些东西，或者不理解 Go 如何处理并发(或者我对并发本身的了解)，我设计了一些代码来理解多个生产者/消费者。

这是代码:

package main

import (
    "fmt"
    "time"
    // "math/rand"
    "sync"
)

var seq uint64 = 0
var generatorChan chan uint64
var requestChan chan uint64

func makeTimestamp() int64 {
    return time.Now().UnixNano() / int64(time.Millisecond)
}

func generateStuff(genId int) {
    var crap uint64
    for {
        crap = <-requestChan
        // <- requestChan
        seq = seq+1
        fmt.Println("Gen ", genId, " - From : ", crap, " @", makeTimestamp())
        generatorChan <- uint64(seq)
    }
}

func concurrentPrint(id int, work *sync.WaitGroup) {
    defer work.Done()

    for i := 0; i < 5; i++ {
        requestChan<-uint64(id)
        fmt.Println("Conc", id, ": ", <-generatorChan)
    }
}

func main() {
    generatorChan = make(chan uint64)
    requestChan = make(chan uint64)
    var wg sync.WaitGroup
    for i := 0; i < 20; i++ {
        go generateStuff(i)
    }
    maximumWorker := 200
    wg.Add(maximumWorker)
    for i := 0; i < maximumWorker; i++ {
        go concurrentPrint(i, &wg)
    }
    wg.Wait()
}

运行时它会打印(主要是按顺序)从 1 到 1000 的所有数字(200 个消费者每人获得一个数字 5 次)。 我原以为有些消费者会打印完全相同的数字，但似乎 requestChan 就像一个屏障一样工作，即使有 20 个 goroutines 服务于 generateStuff通过增加全局变量来生成数字。

一般来说，我对 Go 或并发有什么误解？

我预计会出现类似两个 generateStuff 类型的例程的情况，它们会一起被唤醒并同时增加 seq，从而让两个消费者打印相同的数字 2次。

编辑 playgolang 代码:http://play.golang.org/p/eRzNXjdxtZ

Answer 1

您有多个可以同时运行并同时尝试和发出请求的工作器。由于 requestChan 是无缓冲的，它们都阻塞等待读取器同步并接受它们的请求。

您有多个生成器，它们将通过 requestChan 与请求者同步，产生结果，然后阻塞在未缓冲的 generatorChan 上，直到工作人员读取结果。请注意，它可能是不同的 worker 。

没有额外的同步，所以其他一切都是不确定的。

• 一个生成器可以处理所有请求。
• 生成器可以获取请求并通过递增 seq 在任何其他生成器碰巧有机会运行之前。如果只有一个处理器，这甚至很有可能。
• 所有的生成器都可以获取请求，但最终都希望在同一时间递增 seq，从而导致各种问题。
• 工作人员可以从他们碰巧发送到或来自完全不同的生成器的同一生成器获得响应。

一般来说，如果不添加同步来强制执行其中一种行为，您就无法确保其中任何一种行为确实发生。

请注意，数据竞争本身就是另一个不确定性事件。有可能获得任意值、程序崩溃等。假设在竞争条件下该值可能仅因一个或一些此类相对无害的结果而偏离是不安全的。

对于实验，您最好的办法就是启动 GOMAXPROCS。通过环境变量(例如 env GOMAXPROCS=16 go run foo.go 或 env GOMAXPROCS=16 ./foo after go build ) 或从您的程序中调用 runtime.GOMAXPROCS(16)。默认值为 1，这意味着可能会隐藏数据竞争或其他“奇怪”行为。

您还可以通过在不同点添加对 runtime.Gosched 或 time.Sleep 的调用来稍微影响事情。

如果您使用竞争检测器(例如使用 go run -race foo.goo 或 go build -race)，您还可以看到数据竞争。程序不仅应该在退出时显示“发现 1 个数据争用”，而且还应该在首次检测到争用时转储大量带有堆栈跟踪的详细信息。

这是用于实验的代码的“清理”版本:

package main

import (
    "log"
    "sync"
    "sync/atomic"
)

var seq uint64 = 0
var generatorChan = make(chan uint64)
var requestChan = make(chan uint64)

func generator(genID int) {
    for reqID := range requestChan {
        // If you want to see a data race:
        //seq = seq + 1
        // Else:
        s := atomic.AddUint64(&seq, 1)
        log.Printf("Gen: %2d, from %3d", genID, reqID)
        generatorChan <- s
    }
}

func worker(id int, work *sync.WaitGroup) {
    defer work.Done()

    for i := 0; i < 5; i++ {
        requestChan <- uint64(id)
        log.Printf("\t\t\tWorker: %3d got %4d", id, <-generatorChan)
    }
}

func main() {
    log.SetFlags(log.Lmicroseconds)
    const (
        numGen    = 20
        numWorker = 200
    )
    var wg sync.WaitGroup
    for i := 0; i < numGen; i++ {
        go generator(i)
    }
    wg.Add(numWorker)
    for i := 0; i < numWorker; i++ {
        go worker(i, &wg)
    }
    wg.Wait()
    close(requestChan)
}

Playground (但请注意， Playground 上的时间戳将没有用，调用 runtime.MAXPROCS 可能不会执行任何操作)。进一步注意，playground 缓存结果，因此重新运行完全相同的程序将始终显示相同的输出，您需要进行一些小的更改或只在您自己的机器上运行它。

很多小的变化，比如分流生成器，使用 log 与 fmt 因为前者提供并发保证，消除数据竞争，使输出看起来更好，等等。