Go语言学习笔记（七）杀手锏 Goroutine + Channel

时间 2019-11-06

标签语言学习笔记杀手锏 goroutine channel 繁體版

原文原文链接

Goroutine

Go语言的主要的功能在于使人简易使用的并行设计，这个方法叫作Goroutine，经过Goroutine可以让你的程序以异步的方式运行，而不须要担忧一个函数致使程序中断，所以Go语言也很是地适合网络服务。java

咱们经过go让其中一个函数同步运行，如此就不须要等待该函数运行完后才能运行下一个函数。python

func main() {
    // 经过 `go`，咱们能够把这个函数异步执行，这样就不会阻塞往下执行。
    go loop()
    // 执行 Other
}

Goroutine是相似线程的概念（但Goroutine并非线程）。线程属于系统层面，一般来讲建立一个新的线程会消耗较多的资源且管理不易。而 Goroutine就像轻量级的线程，但咱们称其为并发，一个Go程序能够运行超过数万个 Goroutine，而且这些性能都是原生级的，随时都可以关闭、结束。一个核内心面能够有多个Goroutine，经过GOMAXPROCS参数你可以限制Gorotuine能够占用几个系统线程来避免失控。c++

在内置的官方包中也不时可以看见Goroutine的应用，像是net/http中用来监听网络服务的函数其实是建立一个不断运行循环的Goroutine。安全

设置同时执行的cpu数（GOMAXPROCS）

GOMAXPROCS 在调度程序优化后会去掉，默认用系统全部资源。网络

func main() {
    num := runtime.NumCPU()    //本地机器的逻辑CPU个数
    runtime.GOMAXPROCS(num)    //设置可同时执行的最大CPU数，并返回先前的设置
    fmt.Println(num)
}

Goroutine中使用recover

应用场景，若是某个goroutine panic了，并且这个goroutine里面没有捕获(recover)，那么整个进程就会挂掉。因此，好的习惯是每当go产生一个goroutine，就须要写下recover。并发

var (
    domainSyncChan = make(chan int, 10)
)

func domainPut(num int) {
    defer func() {
        err := recover()
        if err != nil {
            fmt.Println("error to chan put.")
        }
    }()
    domainSyncChan <- num
    
    panic("error....")
}

func main() {
    for i := 0; i < 10; i++ {
        domainName := i
        go domainPut(domainName)
    }
    time.Sleep(time.Second * 2)
}

Goroutine 栗子

package main

import (
    "fmt"
    "sync"
    "time"
)

var (
    m    = make(map[int]uint64)
    lock sync.Mutex //申明一个互斥锁
)

type task struct {
    n int
}

func calc(t *task) {
    defer func() {
        err := recover()
        if err != nil {
            fmt.Println("error...")
            return
        }
    }()

    var sum uint64
    sum = 1
    for i := 1; i < t.n; i++ {
        sum *= uint64(i)
    }

    lock.Lock() //写全局数据加互斥锁
    m[t.n] = sum
    lock.Unlock() //解锁
}

func main() {
    for i := 0; i < 10; i++ {
        t := &task{n: i}
        go calc(t) // Goroutine来执行任务
    }

    time.Sleep(time.Second) // Goroutine异步，因此等一秒到任务完成

    lock.Lock() //读全局数据加锁
    for k, v := range m {
        fmt.Printf("%d! = %v\n", k, v)
    }
    fmt.Println(len(m))
    lock.Unlock() //解锁
}

Goroutine 栗子（等待全部任务退出主程序再退出）

package main

import (
    "sync"
    "fmt"
    "time"
)

func calc(w *sync.WaitGroup, i int)  {
    fmt.Println("calc: ", i)
    time.Sleep(time.Second)
    w.Done()
}

func main()  {
    wg := sync.WaitGroup{}
    for i:=0; i<10; i++ {
        wg.Add(1)
        go calc(&wg, i)
    }
    wg.Wait()
    fmt.Println("all goroutine finish")
}

Channel

channel，管道、队列，先进先出，用来异步传递数据。channel加上goroutine，就造成了一种既简单又强大的请求处理模型，使高并发和线程同步之间代码的编写变得异常简单。dom

线程安全，多个goroutine同时访问，不须要加锁。异步

channel是有类型的，一个整数的channel只能存放整数。函数

channel使用

//chan申明
var userChan chan interface{}          // chan里面放interface类型
userChan = make(chan interface{}, 10)  // make初始化，大小为10

var readOnlyChan <-chan int            // 只读chan
var writeOnlyChan chan<- int           // 只写chan

//chan放取数据
userChan <- "nick"
name := <- userChan
name, ok := <- userChan

//关闭chan
intChan := make(chan int, 1)
intChan <- 9
close(intChan)

// range chan
intChan := make(chan int, 10)
for i := 0; i < 10; i++ {
    intChan <- i
}
close(intChan)

for v := range intChan {
    fmt.Println(v)
}

放入chan数据个数超过初始化指定大小会怎样？

userChan := make(chan interface{})
userChan <- "nick"
// 错误！fatal error: all goroutines are asleep - deadlock!
// 开启race会一直阻塞

开启一个goroutine来放入初始化未指定大小的chan不会报错。高并发

即放即走，在等放入时有来拿数据的，就直接拿走。

userChan := make(chan interface{})
go func() {
    userChan <- "nick"
}()
name := <- userChan

userChan := make(chan interface{})
go func() {
    for {
        userChan <- "nick"
    }
}()
for {
    name := <- userChan
    fmt.Println(name)
    time.Sleep(time.Millisecond)
}

chan关闭与不关闭

关闭chan后再放入数据会 panic: send on closed channel。

chan不关闭取超数据的状况会报 deadlock

func main() {
    intChan := make(chan int, 10)

    for i := 0; i < 10; i++ {
        intChan <- i
    }
    for {
        //十次后 fatal error: all goroutines are asleep - deadlock!
        i := <- intChan
        fmt.Println(i)
        time.Sleep(time.Second)
    }
}

chan关闭的状况取超出值为类型默认值，如int为0

func main() {
    intChan := make(chan int, 10)

    for i := 0; i < 10; i++ {
        intChan <- i
    }
    close(intChan)

    for {
        i := <- intChan
        //十次后i值都为0，不报错
        time.Sleep(time.Second)
        fmt.Println(i)
    }
}

判断chan是否取完

func main() {
    intChan := make(chan int, 10)

    for i := 0; i < 10; i++ {
        intChan <- i
    }
    close(intChan)

    for {
        i, ok := <- intChan
        if !ok {
            fmt.Println("channel is close.")
            return
        }
        fmt.Println(i)
    }
}

channel 栗子

栗子一

func sendData(ch chan<- string) {
    ch <- "go"
    ch <- "java"
    ch <- "c"
    ch <- "c++"
    ch <- "python"
    close(ch)
}

func getData(ch <-chan string, chColse chan bool) {
    for {
        str, ok := <-ch
        if !ok {
            fmt.Println("chan is close.")
            break
        }
        fmt.Println(str)
    }
    chColse <- true
}

func main() {
    ch := make(chan string, 10)
    chColse := make(chan bool, 1)
    go sendData(ch)
    go getData(ch, chColse)
    <-chColse
    close(chColse)
}

栗子二：interface类型chan，取出后转化为对应类型。

type user struct {
    Name string
}

func main() {
    userChan := make(chan interface{}, 1)

    u := user{Name: "nick"}
    userChan <- &u
    close(userChan)

    var u1 interface{}
    u1 = <-userChan

    var u2 *user
    u2, ok := u1.(*user)
    if !ok {
        fmt.Println("cant not convert.")
        return
    }
    fmt.Println(u2)
}

channel 超时处理

利用select来处理chan超时。

for {
    select {
    case v := <-chan1:
        fmt.Println(v)
    case v := <-chan2:
        fmt.Println(v)
    default:
        time.Sleep(time.Second)
        fmt.Println("timeout...")
    }
}

time.After()定时器来作处理。

在time.After()计时器触发以前，底层计时器不会被垃圾收集器回收。

select {
case m := <-c:
    handle(m)
case <-time.After(5 * time.Minute):
    fmt.Println("timed out")
}

定时器栗子

Goroutine+Channel 栗子

栗子一

多个goroutine处理任务；

等待一组channel的返回结果。

func calc(taskChan, resChan chan int, exitChan chan bool) {
    defer func() {
        err := recover()
        if err != nil {
            fmt.Println("error...")
            return 
        }
    }()
    
    for v := range taskChan {
        // 任务处理逻辑
        flag := true
        for i := 2; i < v; i++ {
            if v%i == 0 {
                flag = false
                break
            }
        }
        if flag {
            //结果进chan
            resChan <- v
        }
    }
    //处理完进退出chan
    exitChan <- true
}

func main() {
    //任务chan
    intChan := make(chan int, 1000)
    //结果chan
    resChan := make(chan int, 1000)
    //退出chan
    exitChan := make(chan bool, 8)

    go func() {
        for i := 0; i < 1000; i++ {
            intChan <- i
        }
        close(intChan)
    }()

    //启动8个goroutine作任务
    for i := 0; i < 8; i++ {
        go calc(intChan, resChan, exitChan)
    }

    go func() {
        //等全部goroutine结束
        for i := 0; i < 8; i++ {
            <-exitChan
        }
        close(resChan)
        close(exitChan)
    }()

    for v := range resChan {
        fmt.Println(v)
    }
}

栗子二

等待一组channel的返回结果 sync.WaitGroup 的解决方法。

WaitGroup用于等待一组线程的结束。父线程调用Add方法来设定应等待的线程的数量。每一个被等待的线程在结束时应调用Done方法。同时，主线程里能够调用Wait方法阻塞至全部线程结束。

func merge(cs <-chan int) <-chan int {
    var wg sync.WaitGroup
    out := make(chan int)

    output := func(c <-chan int) {
        for n := range c {
            out <- n
        }
        wg.Done()
    }
    wg.Add(len(cs))

    for _, c := range cs {
        go output(c)
    }

    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}