利用团队协作进行并行工作

本文探讨了利用 Go 通道和工作组模式模拟 CDN 服务器工作人员的问题。代码中存在一个死锁问题，因为 goroutine 正在等待从通道读取，而没有剩余的 goroutine 可以写入。为了解决这个问题，作者建议在完成写入后关闭通道，并添加一个 waitgroup 来等待特定 goroutine 完成。通过这种方法，可以确保所有 goroutine 在继续之前完成各自的任务，从而避免死锁。

尝试使用 go 通道和工作组模式来模拟 cdn 服务器工作人员。

func main() {
    var wg sync.waitgroup
    filejobs := make(chan string)
    compress := make(chan compressionstatus)
    upload := make(chan uploadresult)

// get list of files to distribute from a text file. this is usually 1-1.5 gb, just filenames
    go listfiles("/users/rverma/go/src/github.com/dsa/gopher/images.txt", filejobs)
    go listfiles("/users/rverma/go/src/github.com/dsa/gopher/videos.txt", filejobs)

    go compressfile(filejobs, compress)
    go compressfile(filejobs, compress)
    go compressfile(filejobs, compress)
    go compressfile(filejobs, compress)
    go compressfile(filejobs, compress)
    go compressfile(filejobs, compress)

    wg.add(3)
    go uploadtoall(compress, upload, &wg)
    go uploadtoall(compress, upload, &wg)
    go uploadtoall(compress, upload, &wg)

    wg.wait()
    close(filejobs)
    close(compress)
    close(upload)
}

func listfiles(filename string, c chan<- string) {
    file, err := os.openfile(filename, os.o_rdonly, os.modeperm)
    if err != nil {
        panic("file not found")
    }
    defer file.close()
    r := bufio.newscanner(file)
    for r.scan() {
        c <- r.text()
    }
}

type compressionstatus struct {
    file       string
    compressed string
    status     bool
}

func compressfile(filejob <-chan string, out chan<- compressionstatus) {
    for filename := range filejob {
        fmt.printf("compressing %s\n", filename)
        fib(25) // calculate fibonnaci number, keep cpu busy
        fmt.printf("compressed %s\n", filename)
        out <- compressionstatus{
            file:       filename,
            compressed: filename + ".compressed",
            status:     true,
        }
    }
}

func uploadtoall(comressedfile <-chan compressionstatus, result chan<- uploadresult, wg *sync.waitgroup) {
    for filename := range comressedfile {
        go func() {
            result <- us(filename.compressed)
        }()
        go func() {
            result <- ind(filename.compressed)
        }()
    }
    wg.done()
}

type uploadresult string

type uploadserver func(region string) uploadresult

var (
    us  = fileuploader("us")
    ind = fileuploader("ind")
)

// can't change construct
func fileuploader(region string) uploadserver {
    sleep := time.millisecond * 500
    if region == "ind" {
        sleep *= 4
    }
    return func(filename string) uploadresult {
        fmt.printf("uploading %s to server %s\n", filename, region)
        time.sleep(sleep)
        fmt.printf("upload %s completed to server  %s\n", filename, region)
        return uploadresult(region)
    }
}

虽然代码可以工作，但是当它完成时，它会显示异常，例如。当我们关闭从通道接收的函数，而发送的循环仍然处于活动状态时，这似乎是一个异常。

fatal error: all goroutines are asleep - deadlock!

goroutine 1 [semacquire]:
sync.runtime_Semacquire(0xc00009a004)
        /usr/local/Cellar/go/1.13.3/libexec/src/runtime/sema.go:56 +0x42
sync.(*WaitGroup).Wait(0xc00009a004)
        /usr/local/Cellar/go/1.13.3/libexec/src/sync/waitgroup.go:130 +0x64
main.main()
        /**/file_compressor.go:34 +0x2e1

goroutine 20 [chan receive]:
main.compressFile(0xc000074060, 0xc0000740c0)
        /**/file_compressor.go:75 +0x230
created by main.main
        /**/file_compressor.go:22 +0x13e

有点困惑解决这个问题的最佳方法是什么。 另外，这应该在 8 核机器上运行，我添加了 6 个 go 工作线程进行压缩，作为其 cpu 密集型进程。想知道我们是否可以优化它或使代码更简洁。

解决方案

compressfile 正在等待从 filejob 读取，但没有剩余的 goroutines 可以写入它。 uploadtoall 也在等待 compressedfile，但这不会发生，所以 wg.done 不会被调用。因此，主 goroutine 也在等待 wg.wait。这意味着，所有 goroutine 都在等待某些东西，但没有一个 goroutine 取得进展，因此陷入僵局。

处理此问题的一种方法是在完成写入后关闭通道。这将终止从通道读取的 for 循环。由于您从多个 goroutine 写入同一个通道，因此您并不真正知道何时完成。您可以为 listfiles goroutine 添加一个新的 waitgroup，并等待这两个 goroutine 完成，然后关闭通道。 compress 和 uploadall 组相同。你可以这样做：

wglist.add(2)
go func() {
  wglist.wait()
  close(filejobs)
}()
go listfiles(&wglist,...)
...

不要忘记这一点：

func listFiles(wgList *sync.WaitGroup,...) {
  defer wgList.Done()
  ...
}

对于其他组来说也是类似的......

本篇关于《利用团队协作进行并行工作》的介绍就到此结束啦，但是学无止境，想要了解学习更多关于Golang的相关知识，请关注golang学习网公众号！