如何在工作池中发送 GO 例程

Question

我正在编写一个算法来将图像分解为多个片段并对其进行操作，但是我目前使用 Go 例程的方式并不是很理想。

我想将其拆分为一个工作池，启动例程并让每个工作人员接受新工作，直到完成映像。

我把它分成 8 个这样：

var bounds = img.Bounds()
            var halfHeight = bounds.Max.Y / 2
            var eighthOne = halfHeight / 4
            var eighthTwo = eighthOne + eighthOne
            var eighthThree = eighthOne + eighthTwo
            var eighthFive = halfHeight + eighthOne
            var eighthSix = halfHeight + eighthTwo
            var eighthSeven = halfHeight + eighthThree

            elapsed := time.Now()
            go Threshold(pic, c2, 0, eighthOne)
            go Threshold(pic, c5, eighthOne, eighthTwo)
            go Threshold(pic, c6, eighthTwo, eighthThree)
            go Threshold(pic, c7, eighthThree, halfHeight)
            go Threshold(pic, c8, halfHeight, eighthFive)
            go Threshold(pic, c9, eighthFive, eighthSix)
            go Threshold(pic, c10, eighthSix, eighthSeven)
            go Threshold(pic, c11, eighthSeven, bounds.Max.Y)

然后我一个接一个地启动 Go 例程，如何将其优化为工作系统？

谢谢

Answer 1

这里你有一个通用模式来实现并发图像处理器，让调用者控制图像分区以将工作分成 n 个部分和执行的并发级别（即用于执行 (可能不同）处理作业的数量）。

参见 pprocess 函数，它实现了采用 Partitioner 和 Processor 的整个模式，前者是一个函数，负责返回 n 个图像分区进行操作，后者是一个函数，它将用于处理每个分区。

我在 func splitVert 中实现了您在代码示例中表达的垂直分割，它返回一个可以将图像分割成 n 个垂直部分的函数。

为了做一些实际的工作，我实现了gray func，这是一个Processor，它将像素颜色转换为灰度（亮度）。

这是工作代码：

type MutableImage interface {
    image.Image
    Set(x, y int, c color.Color)
}

type Processor func(MutableImage, image.Rectangle)

type Partitioner func(image.Image) []image.Rectangle

func pprocess(i image.Image, concurrency int, part Partitioner, proc Processor) image.Image {
    m := image.NewRGBA(i.Bounds())
    draw.Draw(m, i.Bounds(), i, i.Bounds().Min, draw.Src)
    var wg sync.WaitGroup
    c := make(chan image.Rectangle, concurrency*2)
    for n := 0; n < concurrency; n++ {
        wg.Add(1)
        go func() {
            for r := range c {
                proc(m, r)
            }
            wg.Done()
        }()
    }
    for _, p := range part(i) {
        c <- p
    }
    close(c)
    wg.Wait()
    return m
}

func gray(i MutableImage, r image.Rectangle) {
    for x := r.Min.X; x <= r.Max.X; x++ {
        for y := r.Min.Y; y <= r.Max.Y; y++ {
            c := i.At(x, y)
            r, g, b, _ := c.RGBA()
            l := 0.299*float64(r) + 0.587*float64(g) + 0.114*float64(b)
            i.Set(x, y, color.Gray{uint8(l / 256)})
        }
    }
}

func splitVert(c int) Partitioner {
    return func(i image.Image) []image.Rectangle {
        b := i.Bounds()
        s := float64(b.Dy()) / float64(c)
        rs := make([]image.Rectangle, c)
        for n := 0; n < c; n++ {
            m := float64(n)
            x0 := b.Min.X
            y0 := b.Min.Y + int(0.5+m*s)
            x1 := b.Max.X
            y1 := b.Min.Y + int(0.5+(m+1)*s)
            if n < c-1 {
                y1--
            }
            rs[n] = image.Rect(x0, y0, x1, y1)
        }
        return rs
    }
}

func main() {
    i, err := jpeg.Decode(os.Stdin)
    if err != nil {
        log.Fatalf("decoding image: %v", err)
    }
    o := pprocess(i, runtime.NumCPU(), splitVert(8), gray)
    err = jpeg.Encode(os.Stdout, o, nil)
    if err != nil {
        log.Fatalf("encoding image: %v", err)
    }
}