Go并发特点

  • goroutine只是由官方实现的超级"线程池"而已,每个实例4-5kb的栈内存占用和用于实现机制而大幅减少的创建和销毁开销。

  • 并发不是并行(多CPU): Concurrency Is Not Parallelism

    SRE实战 互联网时代守护先锋,助力企业售后服务体系运筹帷幄!一键直达领取阿里云限量特价优惠。
  • 并发主要由切换时间片来实现"同时"运行,并行则是直接利用多核实现多线程的运行,但Go可以设置使用核数,以发挥多核计算机的能力。
    • 通过go关键字实现多线程
    package main
    import (
        "fmt"
        "time"
    )
    
    func Go() {
        fmt.Println("1234...") 
    
    }
    
    func main() {
        go Go()                //go关键字构成多线程
        time.Sleep(2 * time.Second) //主程序睡眠2s
    }
  • Goroutine 奉行通过通信来共享内存,而不是共享内存来通信

Channel

  • Channel是goroutine沟通的桥梁,大都是阻塞同步的
  • 通过make创建,close关闭(当程序简单时,回自动关闭)

    package main
    import  "fmt"
    
    func main() {                        //主程序
        c := make(chan bool)           //初始化一个chan类型
        go func() {                 //子程序
            fmt.Println("123...")      //执行主程序
            c <- true                       //通过<-存入bool类型到chan中
        }()
    fmt.Println(1)                  //程序执行步骤:1st
    read_chan := <-c                         //<-c 从chan中读取bool,程序执行步骤:2nd
    fmt.Println(read_chan)                 //程序执行步骤:3rd
    }
    
    /*output
    1st     1           
    2nd     123...
    3rd     true
    */

    注意以上程序的执行顺序(channel无缓存时):先执行读取操作c<-c,因为channel中没有值,所以程序发生阻塞,此时执行chanel写操作,然后再执行读操作。

  • Channel是引用类型

  • 可以使用for range来迭代不断操作channel

    package main
    import  "fmt"
    
    func main() {
        c := make(chan bool)       //初始化一个chan类型
        go func() {                        //go结合匿名函数,构造并发
            fmt.Println("123...")     //执行主程序
            c <- true                     //通过<-存入bool类型到chan中
            close(c)                      //关闭通道:必须明确在哪个地方关闭
        }()
    
        for v := range c {       //for循环chanel
        }
    }
    
    /*output
        123...
        true
    */
  • 可以设置单向(读&写)或双向通道--默认是双向通道

  • 可以设置缓存大小(默认为0,阻塞),在未被填充前不会发生阻塞(异步),比如缓存20个,可以同时进行20个读操作或者写操作,注意读的操作先于写的操作

    package main
    import (
        "fmt"
    )
    
    func main() {                                       //主程序
        c := make(chan bool, 1)             //初始化一个chan类型,缓存为2
        go func() {                                  //子程序
            fmt.Println("123...")            //执行主程序,执行步骤:2
            c <- true                            //写操作,执行步骤:2
        }()
        fmt.Println(2)                             //执行步骤:1
        fmt.Println(123, <-c)                 //读操作,执行步骤:2
        fmt.Println(3)                            //执行步骤:3
    }
    
    /*output
    1   2
    2   123...
    2   123 true
    3   3
    */
  • 并行并发,利用多核CPU,当使用单线程执行的时候,就是同步按部就班的执行程序,使用多核时,且没有设置channel缓存机制时是随机异步发生的。但是这样就会造成一个问题,因为是随机的,所以CPU在选择的时候,某有一个程序并没有执行,会造成程序执行的遗漏。

    package main
    import (
        "fmt"
        "runtime"
    )
    
    func main() {                                                                        //主程序
        runtime.GOMAXPROCS(runtime.NumCPU())              //获取CPU的核数
        c := make(chan bool)                                                  //创建channel
        for i := 0; i < 10; i++ {                                                  //启动10次
            go Go(c, i)
        }
        <-c                                                                             //读取chanel值
    }
    
    func Go(c chan bool, index int) {
        a := 1
        for i := 0; i < 1000000; i++ {
            a += i
        }
        fmt.Println(index, a)
    
        if index == 9 {
            c <- true //向chanel传入值
        }
    }
    
    /*output
    4 499999500001
    9 499999500001
    */
  • 解决异步并发数据丢失:方式1:根据并发执行次数为channel设置同等数量的缓存机制。

    package main
    
    import (
        "fmt"
        "runtime"
    )
    
    func main() {                                                             //主程序
        runtime.GOMAXPROCS(runtime.NumCPU())   //获取CPU的核数
        c := make(chan bool, 10)                                //创建channel
        for i := 0; i < 10; i++ {                                      //启动10次
            go Go(c, i)
        }
        for i := 0; i < 10; i++ {                                      //for 循环为chan循环读取十次
            <-c                                                         //读取chanel值
        }
    }
    
    func Go(c chan bool, index int) {
        a := 1
        for i := 0; i < 1000000; i++ {
            a += i
        }
        fmt.Println(index, a)
    
        c <- true                                           //向chanel传入值
    
    }
    
    /*output
        0 499999500001
        9 499999500001
        1 499999500001
        5 499999500001
        6 499999500001
        2 499999500001
        7 499999500001
        3 499999500001
        8 499999500001
        4 499999500001
    */
  • 解决方法2:使用同步 sync 包, sync.WaitGroup{} 创建任务池;sync.add(n)为任务池添加任务个数; sync.Wait()主程序等待(守护进程); sync.Done()子进程全部执行完毕告知主进程,退出程序。

    package main
    import (
        "fmt"
        "runtime"
        "sync"
    )
    
    func main() { //主程序
        runtime.GOMAXPROCS(runtime.NumCPU()) //获取CPU的核数
        wg := sync.WaitGroup{}               //调用同步方法,类似一个池子
        wg.Add(10)                           //增加10个任务数
        for i := 0; i < 10; i++ {            //启动10次
            go Go(&wg, i)                    //wg类型是值拷贝,所以调用时,使用指针传递
        }
        wg.Wait()                           //等待10个子进程全部执行完毕后,退出
        fmt.Println("Main process exit!")
    }
    
    func Go(wg *sync.WaitGroup, index int) {
        a := 1
        for i := 0; i < 1000000; i++ {
            a += i
        }
        fmt.Println(index, a)
        wg.Done()                          //子程序退出
    }
    /* 输出
        0 499999500001
        9 499999500001
        1 499999500001
        5 499999500001
        6 499999500001
        2 499999500001
        7 499999500001
        3 499999500001
        8 499999500001
        4 499999500001
        Main process exit!
    */

Select

  • 可处理一个或多个channel的发送与接收

    package main
    import (
        "fmt"
    )
    
    func main() { //主程序
        c1, c2 := make(chan int), make(chan string) //创建两个channel
        o := make(chan bool)                        //创建一个信号通道,用来监控通道c1,c2
        go func() {
            fmt.Println("start...") //go 匿名函数
            for {                   //这是一个无线循环
                select { //类似switch语句,进行判断
                case v, ok := <-c1: //读取通道c1的值,赋值给变量v
                    if !ok { //如果从chan,c1中未读取到值,发生阻塞
                        o <- true //在信号通道O中写入true,程序break
                        break
                    }
                    fmt.Println("c1", v)
    
                case v, ok := <-c2: //v判断从通道2中读取到的值
                    if !ok { //如果读取失败
                        o <- true //向信号通道中传递一个信号
                        break     //程序break
                    }
                    fmt.Println("c2", v)
                }
            }
        }()
    
        c1 <- 1 //通道c1中写入数据
        c2 <- "mm"
        c1 <- 2
        c2 <- "my"
        close(c1) //通道操作完后需关闭通道
        close(c2)
        <-o //最后读取信号通道内容
    }
    /*输出
    start...
    c1 1
    c2 mm
    c1 2
    c2 my
    */

    当有多个channel时,是无法对其中多个channel进行关闭的,只能判断其中一个channel是否关闭,只能进行某个channel进行关闭

  • 同时又多个可用的channel时按随机顺序处理

    package main
    import (
        "fmt"
    )
    
    func main() { //主程序
        c := make(chan int)
        go func() { //go匿名函数
            for v := range c { //循环读取出chan的值
                fmt.Println(v)
            }
        }()
    
        for {
            select { //select判断
            case c <- 0:
            case c <- 1:
            }
        }
    }
    
    /*输出
    0
    1
    1
    1
    0
    ...
    */
  • 可用空的select来阻塞main函数--应用(GUI程序)

    package main
    import  "fmt"
    
    
    func main() { //主程序
        c := make(chan int)
        go func() { //go匿名函数
            for v := range c { //循环读取出chan的值
                fmt.Println(v)
            }
        }()
    
        for {
            select { //空select可阻塞main主函数
            }
        }
    }
  • 可设置超时

    package main
    import (
        "fmt"
        "time"
    )
    
    func main() { //主程序
        c := make(chan int)
        select {
        case v := <-c: //channal中没有内容,所以会执行下一个case
            fmt.Println(v)
        case <-time.After(3 * time.Second): //selesct设置超时时间,
            fmt.Println("Timeout!")
        }
    
    }
    
    /*输出
    Timeout
    */
扫码关注我们
微信号:SRE实战
拒绝背锅 运筹帷幄