Go语言学习笔记　－　并发

Goroutine

　　Go在语言层面对并发编程提供支持，采用轻量级线程（协程）实现。只需要在函数调用语句前添加go关键字，就可以创建并发执行单元。开发人员无需了解任何执行细节，调度器会自动将其安排到合适的系统线程上执行。goroutine是一种非常轻量级的实现，可在单个进程里执行成千上万的并发任务。事实上，入口函数main就以goroutine运行。另有与之配套的channel类型，用以实现“以通讯来共享内存”的CSP模式。

go func() {
    println("Hello, World!")
}

　　调度器不能保证多个goroutine执行次序，且进程退出时不会等待它们结束。默认情况下，进程启动后仅允许一个系统线程服务于goroutine。可使用环境变量或标准函数runtime.GOMAXPROCS修改（Go 1.5默认方式）。让高度器用多个线程实现多核并行，而不仅仅是并发。

func sum(id int) {
    var x int64
    for i := 0; i < math.MaxUint32; i++ {
        x += int64(i)
    }
    println(id, x)
}
func main() {
    wg := new(sync.WaitGroup)
    wg.Add(2)
    for i := 0; i < 2; i++ {
        go func(id int) {
            defer wg.Done()
            sum(id)
        }(i)
    }
    wg.Wait()
}
输出:
$ go build -o test
$ time -p ./test
0 9223372030412324865
1 9223372030412324865
real 7.70 // 程序开始到结束时间差 (非非 CPU 时间)
user 7.66 // 用用户态所使用用 CPU 时间片片 (多核累加)
sys 0.01 // 内核态所使用用 CPU 时间片片
$ GOMAXPROCS=2 time -p ./test
0 9223372030412324865
1 9223372030412324865
real 4.18
user 7.61// 虽然总时间差不多,但由 2 个核并行行,real 时间自自然少了许多。
sys 0.02

　　调用 runtime.Goexit 将立即终止当前 goroutine 执行,调度器确保所有已注册 defer延迟调用被执行。

func main() {
    wg := new(sync.WaitGroup)
    wg.Add(1)
    go func() {
        defer wg.Done()
        defer println("A.defer")
        func() {
            defer println("B.defer")
            runtime.Goexit() // 终止止当前 goroutine
            println("B") // 不会执行行
        }()
        println("A")    // 不会执行行
    }()
    wg.Wait()
}
//输出:
B.defer
A.defer

　　和协程 yield 作用类似,Gosched 让出底层线程,将当前 goroutine 暂停,放回队列等待下次被调度执行。

func main() {
    wg := new(sync.WaitGroup)
    wg.Add(2)
    go func() {
        defer wg.Done()
        for i := 0; i < 6; i++ {
            println(i)
            if i == 3 { runtime.Gosched() }
        }
    }()
    go func() {
        defer wg.Done()
        println("Hello, World!")
    }()
    wg.Wait()
}
//输出
$ go run main.go
0
1
2
3
Hello, World!
4
5

Channel

　　引用类型Channel是CSP模型的具体实现，用于多个goruntine之间进行通讯。其内部实现了同步，确保了并发安全。默认为同步模式，需要发送和接收配对。否则会被阻塞，直到另一方准备好后被唤醒。

func main() {
    data := make(chan int) // 数据交换队列
    exit := make(chan bool) // 退出通知
    go func() {
        for d := range data {// 从队列迭代接收数据,直到 close 。
            fmt.Println(d)
        }
        fmt.Println("recv over.")
        exit <- true// 发出退出通知。
    }()
    data <- 1// 发送数据。
    data <- 2
    data <- 3
    close(data)// 关闭队列。
    fmt.Println("send over.")
    <-exit// 等待退出通知。
}
//输出:
1
2
3
send over.
recv over.

　　异步方式通过判断缓冲区来决定是否阻塞。如果缓冲区已满，发送被阻塞；缓冲区为空，接收被阻塞。通常情况下，异步channel可减少排队阻塞，具备更高的效率。但应该考虑使用指针规避大对象拷贝，将多个元素打包，减少缓冲区大小等。

func main() {
    data := make(chan int, 3)// 缓冲区可以存储 3 个元素
    exit := make(chan bool)
    data <- 1// 在缓冲区未满前,不会阻塞。
    data <- 2
    data <- 3
    go func() {
        for d := range data {// 在缓冲区未空前,不会阻塞。
            fmt.Println(d)
        }
        exit <- true
    }()
    data <- 4// 如果缓冲区已满,阻塞。
    data <- 5
    close(data)
    <-exit
}

　　缓冲区是内部属性,并非类型构成要素。

var a, b chan int = make(chan int), make(chan int, 3)

　　除用用 range 外,还可用 ok-idiom 模式判断 channel 是否关闭。

for {
    if d, ok := <-data; ok {
        fmt.Println(d)
    } else {
        break
    }
}

　　向 closed channel 发送数据引发 panic 错误,接收立即返回零值。而 nil channel,无论收发都会被阻塞。内置函数 len 返回未被读取的缓冲元素数量,cap 返回缓冲区大小。

单向

　　可以将channel隐式转换为单身队列，只收或只发。

c := make(chan int, 3)
var send chan <- int = c // send only
var recv <- chan int = c // receiver only

选择

　　如果需要同时处理多个channel，可以用select语句，它随机选择一个可用的channel做收发操作，或执行default case。

func main() {
    a, b := make(chan int, 3), make(chan int)
    go func() {
        v, ok, s := 0, false, ""
        for {
            select {// 随机选择可用用 channel,接收数据。
            case v, ok = <-a: s = "a"
            case v, ok = <-b: s = "b"
            }
            if ok {
              fmt.Println(s, v)
            } else {
                os.Exit(0)
            }
        }
    }()
    for i := 0; i < 5; i++ {
          select {// 随机选择可用用 channel,发送数据。
          case a <- i:
          case b <- i:
          }
    }
    close(a)
    select {}// 没有可用用 channel,阻塞 main goroutine。
}
//输出:
b 3
a 0
a 1
a 2
b 4

模式

　　用简单工厂模式打包并发任务和channel。

func NewTest() chan int {
    c := make(chan int)
    rand.Seed(time.Now().UnixNano())
    go func() {
        time.Sleep(time.Second)
        c <- rant.Int()
    }()
    return c
}
func main() {
    t := NewTest()
    println(<-t) //等待gorountime结束返回。
}

　　用channel实现信号量(semaphore)。

func main() {
    wg := sync.WaitGroup{}
    wg.Add(3)
    sem := make(chan int, 1)
    for i := 0; i < 3; i++ {
        go func(id int) {
            defer wg.Done()
            sem <- 1 // 向 sem 发送数据,阻塞或者成功。
            for x := 0; x < 3; x++ {
                fmt.Println(id, x)
            }
            <- sem // 接收数据,使得其他阻塞 goroutine 可以发送数据。
        }(i)
    }
    wg.Wait()
}

　　用closed channel发出退出通知。

func main() {
    var wg sync.WaitGroup
    quit := make(chan bool)
    for i := 0; i < 2; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            task := func() {
                println(id, time.Now().Nanosecond())
                time.Sleep(time.Second)
            }
            for {
              select {
              case <- quit: // closed channel 不会阻塞,因此可用作退出通知。
                  return
              default://执行正常任务
                  task()
              }
          }
        }(i)
    }
    time.Sleep(time.Second * 5) // 让测试 goroutine 运行一会。
    close(quit) // 发出退出通知。
    wg.Wait()
}

　　用select 实现超时 (timeout)。channel 是第一类对象,可传参 (内部实现为指针) 或者作为结构成员。

type Request struct {
    data []int
    ret chan int
}
func NewRequest(data ...int) *Request {
    return &Request{ data, make(chan int, 1) }
}
func Process(req *Request) {
    x := 0
    for _, i := range req.data {
        x += i
    }
    req.ret <- x
}
func main() {
    req := NewRequest(10, 20, 30)
    Process(req)
    fmt.Println(<-req.ret)
}

有疑问加站长微信联系（非本文作者）