Goroutine学习笔记(一)

Goroutine

在Go语言中，可以通过go func()来创建一个Goroutine，但是该Goroutine并不会立刻执行

goroutine-without-wait.go

package main

import (
    "fmt"
)

func main() {
    go func() {
        fmt.Printf("world!\n")
    }() //创建一个Goroutine
    fmt.Printf("Hello ")
}

输出结果为

Hello

main()函数执行完毕后，会立刻退出，并不会等待后台Goroutine的执行，为了使Goroutine能够顺利执行，可以在main()函数中调用time.Sleep()，使得Goroutine能够获取CPU时间片并执行

wait-for-goroutine.go

package main

import (
    "fmt"
    "time"
)

func main() {
    go func() {
        fmt.Printf("world!\n")
    }()
    fmt.Printf("Hello ")
    time.Sleep(1 * time.Second) //暂停1s，等待Goroutine执行完毕
}

输出结果为

Hello world!

在main()函数中通过调用time.Sleep()将当前程序进入等待状态，这样Goroutine就可以顺利执行，这样的弊端就是可能等待的时间过长或过短，可以使用Channel等待Goroutine的执行

wait-goroutine-by-channel.go

package main

import (
    "fmt"
)

func main() {
    var c = make(chan struct{})
    go func() {
        fmt.Printf("Hello ")
        c <- struct{}{} //发送一个消息到Channel 或者 <-c
    }()
    <-c //从Channel接受消息 或者  c <- struct{}{}
    fmt.Printf("world!\n")
}

在main()函数中，接收Channel的操作会一直阻塞，直到从Channel中接收到消息才会继续向下执行
如果在main()函数中发送一条消息Channel，那么在其它的Groutine从Channel中接收到该消息之前，程序也会阻塞

GO语言中还提供了一种方式等待Goroutine的执行

wait-goroutine-by-waitGroup.go

package main

import (
    "fmt"
    "sync"
)

func main() {
    var n sync.WaitGroup
    n.Add(1) //加1
    go func() {
        fmt.Printf("Hello ")
        n.Done()
        //相当于减1，一种常见的用法是在func()内部语句开头使用  defer n.Done()
    }()
    n.Wait()//值为0才会向下执行
    fmt.Printf("world!\n")
}

在使用sync.WaitGroup时需要注意n.Add(1)与n.Done()的执行次数需一致，这样可以避免死锁，在循环中一种常用的用法

wait-goroutine-by-waitGroup-forLoop.go

package main

import (
    "fmt"
    "sync"
)

func main() {
    var n sync.WaitGroup
    for i := 0; i < 10; i++ {
        n.Add(1)
        go func(i int) {
            defer n.Done()
            fmt.Printf("Task [%d] Work!\n", i)
        }(i)
    }
    n.Wait()
    fmt.Printf("All task has done!\n")
}

使用Buffer Channel也可以完成此类操作，仍需注意Channel缓冲区的数量要与创建Goroutine的数量一致

wait-goroutine-by-bufferChannel-forLoop.go

package main

import (
    "fmt"
)

func main() {
    var c = make(chan struct{}, 10)
    for i := 0; i < 10; i++ {
        go func(i int) {
            fmt.Printf("Task [%d] word!\n", i)
            c <- struct{}{}
        }(i)
    } //创建10个Goroutine
    
    //等待所有的Goroutine执行完毕
    for j := 0; j < 10; j++ {
        <-c
    } //可以更改J的值看看会发生什么

    fmt.Printf("All task has done!\n")
}

Buffer Channel在缓冲区满时，发送操作会阻塞，在缓冲区为空时，接收操作会阻塞

并行循环

在处理一些矩阵问题时，可以采用矩阵分块的思想利用Goroutine加速实现任务

《Go程序设计语言》第八章有一个关于并行循环的习题，大致是画一幅1024 x 1024像素的图形，能否利用Goroutine对其进行加速？

在刚开始看到题目时，我以为是在调用img.set()时创建Goroutine对其进行加速，于是就有了下面这样的代码

var n sync.WaitGroup
img := image.NewRGBA(image.Rect(0, 0, width, height))
for py := 0; py < height; py++ {
    y := float64(py)/height*(ymax-ymin) + ymin
    for px := 0; px < width; px++ {
        x := float64(px)/width*(xmax-xmin) + xmin
        z := complex(x, y)
        n.Add(1)
        go func(px, py int, z color.Color) {
            defer n.Done()
            img.Set(px, py, z) //将颜色写入对应的位置
        }(px, py, mandelbrot(z))
    }
}
n.Wait() //等待所有Goroutine执行完毕

结果在实际运行时发现比不使用Goroutine还慢，在这段代码中总共创建了1024 x 1024个Goroutine，系统需要花费很长时间用于Goroutine的调度

在看完别人的解答后发现思路是将一个1024 x 1024的矩阵分为y个 (1024 / y) x 1024的矩阵，每个Goroutine负责执行一部分

for i := 0; i < 4; i++ {
    wg.Add(1)
    go func(i int) {
        defer wg.Done()
        //每个Goroutine根据i的值负责一部分区域的颜色填充
        for py := i * 256; py < (i+1)*256; py++ {
            y := float64(py)/height*(ymax-ymin) + ymin
            for px := 0; px < width; px++ {
                x := float64(px)/width*(xmax-xmin) + xmin
                z := complex(x, y)
                img.Set(px, py, mandelbrot(z)) //颜色填充
            }
        }
    }(i)
}
wg.Wait()

超时处理

假设有一个Master节点负责维护任务的状态，其它Worker节点负责执行任务，很多时候对于每个任务都会设置超时处理，如果某个任务没有在规定时间内完成，将任务重新分配给Worker执行

每当Master节点分配一个任务时，可以创建一个Goroutine负责监控该任务的执行状态

下面给出一段Master的伪码

master-with-timeout.go

const (
    UNASSIGN  = 0 //未分配
    INPROCESS = 1 //正在执行
    COMPLETED = 2 //执行完毕
)

func assignWorkRPC() {
    for index := 0; index < len(task); index++ {
        //如果该任务没有被分配，分配该任务
        if task[index] == UNASSIGN {
            task[index] = INPROCESS  //任务状态未正在执行
            go func(index int) {
                //休眠一段时间后检查任务状态
                time.Sleep(10 * time.Second)
                if task[index] != COMPLETED { //任务未在规定时间内完成
                    task[index] = UNASSIGN //任务状态转换为未分配
                }
            }(index)
        }
    }
}

func workCompleteRpc() {
    task[args.TaskId] = COMPLETED //如果Worker执行完任务改变任务状态
}

在处理超时问题时，使用Goroutine与time.sleep()是一种很常见的方法，首先让Goroutine休眠一段时间，接着检查任务的状态，如果任务在规定时间内完成就不做任何改动，如果任务没有在规定时间内完成，需要改变任务的状态