golang学习随便记10

Posted 2021-11-23 sjg20010414

goroutine 和 channel （2）

并发的循环

书中举了一个容易并行处理的问题，就是给一堆图片生成缩略图。每个缩略图都只和自己那个原始图有关，生成缩略图的工作相互独立，所以，这是一个容易并行的问题。但以下代码是错误的：

func makeThumbnails2(filenames []string) {
    for _, f := range filenames {
        go thumbnail.ImageFile(f) // NOTE: ignoring errors
    }
}

上面的 for 循环，为每一个图片处理创建一个 goroutine，结果，函数很快返回了，却不知道那些 goroutine 有没有干完活（事实上大概率没干完）。要避免这种情况，就是使用一个共享的 channel。在这个应用场景中，我们确切知道有几个（len(filenames)）图片在处理，可以通过共享channel在外部进行计数来确认。

func makeThumbnails3(filenames []string) {
    ch := make(chan struct{})
    for _, f := range filenames {
        go func(f string) {        // 注意，在循环中，必须把 f 拷贝给函数闭包的参数 f
            thumbnail.ImageFile(f) // NOTE: ignoring errors
            ch <- struct{}{}        // 每处理一个，发送一个 “信号”
        }(f)
    }
    // Wait for goroutines to complete.
    for range filenames {
        <-ch                        // 接收 “信号”
    }
}

上面的代码的缺点是没有考虑生成缩略图时的错误。下面的代码尝试把发送的 “信号” 从“无类型”信号改成 error

func makeThumbnails4(filenames []string) error {
    errors := make(chan error)

    for _, f := range filenames {
        go func(f string) {
            _, err := thumbnail.ImageFile(f)
            errors <- err                        // 发送 error 类型的 信号
        }(f)
    }

    for range filenames {
        if err := <-errors; err != nil {        // 发生了错误，函数返回
            return err                     // NOTE: incorrect: goroutine leak!
        }
    }

    return nil
}

上述程序在遇到第一个非nil的error时，函数返回了，这样就会造成 errors 这个 channel 没有排空，这样剩下还在工作中的那些 goroutine 试图继续往 errors 灌，而此时不再有接收者了（后一个for因为函数返回已经不再做接收工作），就会造成阻塞（那些goroutine无法继续向errors灌却仍努力灌）。这称为 goroutine 泄漏，可能造成整个程序卡住或out of memory。

我们用带缓冲的 channel 来避免阻塞。同时，下面函数还记录了所有缩略图的文件名。

func makeThumbnails5(filenames []string) (thumbfiles []string, err error) {
    type item struct {
        thumbfile string
        err       error
    }

    ch := make(chan item, len(filenames))                   // 带缓冲，队列大小和文件个数一致
    for _, f := range filenames {
        go func(f string) {
            var it item
            it.thumbfile, it.err = thumbnail.ImageFile(f)
            ch <- it                               // 发送自定义类型，同时包含文件名和 error值
        }(f)
    }

    for range filenames {
        it := <-ch                    // 接收，同时包含了缩略图文件名和error值
        if it.err != nil {
            return nil, it.err
        }
        thumbfiles = append(thumbfiles, it.thumbfile)
    }

    return thumbfiles, nil
}

下面的版本能计算得到所有缩略图文件的总字节数，这比缩略图文件名更有用，因为通常缩略图文件名往往按固定规则从原文件名生成。另外，前面的函数调用时，文件名是直接用函数参数传入的，而这一版本文件名是从 channel 接收得到，而且，每个缩略图的文件大小，也是通过 channel 传递出去的，这样，文件名、处理、大小计算三者是可以解耦的或者说并行的，不过为了完成任务，我们需要知道最后一个 goroutine 什么时候结束（最后结束的不一定是最后启动的）。

sync.WaitGroup 计数器可以帮我们完成任务：启动 goroutine 前计数加1，每个goroutine处理完（包括出错而完）都减1，一直等到计数为0，说明全部完了，可以计算总字节数。

func makeThumbnails6(filenames <-chan string) int64 {
    sizes := make(chan int64)
    var wg sync.WaitGroup         // number of working goroutines 计数器
    for f := range filenames {
        wg.Add(1)                // 加1，表示多了一个工作 goroutine
        // worker
        go func(f string) {
            defer wg.Done()                        // defer 方式确保一个 goroutine 完了减1
            thumb, err := thumbnail.ImageFile(f)
            if err != nil {
                log.Println(err)
                return
            }
            info, _ := os.Stat(thumb) // OK to ignore error
            sizes <- info.Size()                   // 用 sizes channel 发送文件大小
        }(f)
    }

    // closer
    go func() {
        wg.Wait()                        // 阻塞等待计数器到0
        close(sizes)                    // 关闭 sizes 这个 channel，表示不再有文件大小发送
    }()

    var total int64
    for size := range sizes {            // 从 sizes 这个 channel 逐步取出所有文件大小
        total += size
    }
    return total
}

我们需要仔细体会这样的代码。如果没有 golang 的这些设施，我们就必须一个个处理文件，每处理一个把文件大小加一个，现在，不仅处理是并行的，而且，文件总大小计算，并不是等所有文件都处理完才计算的，而是处理过程中，每处理完一个，文件大小就累加上去了。代码能正常工作，是因为最后一个for 在 sizes 这个 channel 上用 range 迭代，它要结束，必须 sizes 被关闭，即 closer 那个 goroutine 应该结束，而 closer 要结束，计数器必须到0，即图片已经被处理完了（虽然我们无法知道它们按怎样的先后顺序被处理完毕的）。其中，Done方法等价于Add(-1)。这个程序的做法，是想使用并发循环，又不知道迭代次数时的通行做法。