Golang学习———多线程编程

2020-07-21

介绍

并发指在同一时间内可以执行多个任务。并发编程含义比较广泛，包含多线程编程、多进程编程及分布式程序等。本章讲解的并发含义属于多线程编程。
Go 语言通过编译器运行时（runtime），从语言上支持了并发的特性。Go 语言的并发通过 goroutine 特性完成。goroutine 类似于线程，但是可以根据需要创建多个 goroutine 并发工作。goroutine 是由 Go 语言的运行时调度完成，而线程是由操作系统调度完成。

Go 语言还提供 channel 在多个 goroutine 间进行通信。goroutine 和 channel 是 Go 语言秉承的 CSP（Communicating Sequential Process）并发模式的重要实现基础。本章中，将详细为大家讲解 goroutine 和 channel 及相关特性。

进程时代

现代化的计算机有了操作系统，每个程序都是一个进程，但是操作系统在一段时间只能运行一个进程，直到这个进程运行完，才能运行下一个进程，这个时期可以成为单进程时代。但由于其速度还是太慢了，比如进程要读数据阻塞了，CPU就在哪浪费着，于是操作系统就具有了最早的并发能力：多进程并发，当一个进程堵塞的时候，就切换到另外等待执行的进程，这样就尽量把CPU利用了起来。
进程的五种状态

线程时代

但后来人们发现进程拥有太多资源，在创建、切换和销毁的时候，都会占用很长的时间，CPU虽然利用起来了，但CPU有很大的一部分都被用来进行进程调度了。
后来操作系统支持了线程，线程在进程里面，运行起来所需要的资源比进程少很多。一个进程可以有多个线程，CPU在执行调度的时候切换的是线程，这很快就能完成。但线程的设计本身有点复杂，而且由于需要考虑很多底层细节，比如锁和冲突检测。
线程池执行流程

协程

但是在当今互联网高并发场景下，因为会消耗大量的内存（每个线程的内存占用级别为MB），线程多了之后调度也会消耗大量的CPU。
线程分为内核态线程和用户态线程，用户态线程需要绑定内核态线程，CPU并不能感知用户态线程的存在，它只知道它在运行1个线程，这个线程实际是内核态线程。用户态线程实际有个名字叫协程（co-routine），为了容易区分，我们使用协程指用户态线程，使用线程指内核态线程。

goroutine

Go为了提供更容易使用的并发方法，使用了goroutine和channel，goroutine是一个轻量级的线程，执行时只需要4-5k的内存，比线程更易用，更高效，更轻便，调度开销比线程小，可同时运行上千万个并发。
go语言中开启一个goroutine非常简单，go函数名（），就开启了个线程。默认情况下，调度器仅使用单线程，要想发挥多核处理器的并行处理能力，必须调用runtine.GOMAXPROCS(n)来设置可并发的线程数，也可以通过设置环境变量GOMAXPROCS打到相同的目的。

runtime 包下提供的与goroutine相关的函数

Gosched()函数
这个函数的作用是让当前goroutine让出CPU，好让其它的goroutine获得执行的机会。同时，当前的goroutine也会在未来的某个时间点继续运行。请看下面的例子：

 package main

import (
    "runtime"
    "fmt"
)

func main(){
    go sayHello()     //启动线程1
    go sayWorld()     // 启动线程2
    var str string    // 等待输入，在这里用来阻止主线程关闭
    fmt.Scan(&str)
}

// 线程1打印 hello
func sayHello(){
    for a:= 0; a < 5; a++ {
        fmt.Print("hello ")
        runtime.Gosched()
        }
}

// 线程1打印 hello
func sayWorld(){
    for a:= 0; a < 5; a++ {
        fmt.Println("world ")
        runtime.Gosched()
        }
}

运行结果如图所示

从以上我们能够看出来，一个线程输出一句后调用Gosched函数，放弃了CPU执行权限，另一个线程获取权限，然后输出内容。就这样两个线程交替获得权限，才输入以上结果。

1 2	var str string // 等待输入，在这里用来阻止主线程关闭 fmt.Scan(&str)

这两句代码是等待输入的意思，在这里用来阻止主线程关闭的。如果没有这两句的话，会发现我们的程序瞬间就结束了，而且什么都没有输出。这是因为主线程关闭之后，所有开启的goroutine都会强制关闭，他还没有来得及输出，就结束了。

运行结果

runtime.NumCPU()
runtime.NumCPU()返回了cpu核数，runtime.NumGoroutine()返回当前进程的goroutine线程数。即便我们没有开启新的goroutine
1
2
3
4
5
6
7
8
9
10
package main

import (
"runtime"
"fmt"
)
func main(){
fmt.Println(runtime.NumCPU())
fmt.Println(runtime.NumGoroutine())
}
运行结果如下

运行结果

Goexit()函数
runtime.Goexit()函数用于终止当前的goroutine，单defer函数将会继续被调用

package main

import (
    "runtime"
    "fmt"
)

func test(){
    defer func(){
        fmt.Println(" in defer")
    }()
    for i := 0; i < 10; i++{
        fmt.Print(i)
        if i > 5{
            runtime.Goexit()
        }
    }
}

func main(){
    go test()
    var str string
    fmt.Scan(&str)
}

运行结果如下

输出结果如下

channel

如果有一种机制，在子线程结束的时候通知一下主线程，然后主线程再关闭，岂不是更好，这样就不用无休止的等待了,于是就有了channel。
goroutine之间通过channel来通讯，可以认为channel是一个管道或者先进先出的队列。你可以从一个goroutine中向channel发送数据，在另一个goroutine中取出这个值。通俗点儿解释就是channel可以在两个或者多个goroutine之间传递消息。在Go中，goroutine和channel是并发编程的两大基石，goroutine用来执行并发任务，channel用来在goroutine之间来传递消息。

引入一个简单的例子


package main

import (
  "fmt"
)

func goroutineA(a <-chan int) {
  for {
    select {
    case val := <-a:
      fmt.Println(val)
    }
  }
}

func main() {
  ch := make(chan int)
  go goroutineA(ch)
  ch <- 3
  ch <- 5
}

这是一段很简单的程序，初始化了一个非缓冲的channel，然后并发一个协程去接受channel中的数据，然后往channel中连续发送两个值。

输出结果如下图所示：
运行结果

使用 make 创建 channel

首先大家先理解一组概念，什么是非缓冲型channel和缓冲型channel？对，其实很简单，make时如果channel空间不为0，就是缓冲型的channel。

// 不带缓冲区
var channel chan int = make(chan int)
// 或
channel := make(chan int)
// 带缓冲区  
ch1 := make(chan int, 10)  
```  

* 那么有缓冲和没有缓冲的区别是什么呢？  
打个比喻，无缓冲的就是一个送信人去你家门口送信 ，你不在家他不走，你一定要接下信，他才会走，无缓冲保证信能到你手上；有缓冲的，就是一个送信人去你家仍到你家的信箱转身就走 ，除非你的信箱满了他必须等信箱空下来，有缓冲的保证信能进你家的邮箱 


* 如果我们将go goroutineA(ch)这行代码往下移，会发生什么？  

> fatal error: all goroutines are asleep - deadlock!  

会报错，因为channel没有缓冲，也没有正在等待接收的goroutine

另外，我们会看到goroutineA的入参是a <-chan int，代表一个只能用于接收的channel，也就是单向channel。大家一定要清楚接收和发送的概念  
chan作为函数返回值的方式有3种:（chan int）、（<- chan int）、（chan <- int），分别代表（可读可写的管道）、（只读管道）、（只写管道），只读管道不能close()，只写管道可以close()

```go
var a <-chan int//单向channel，只用于接收，也就是从channel读取据
var a chan<- int//单向channel，只用于发送，也就是往channel写入数据

引入一个较为复杂的例子

package main

import (
  "fmt"
  "os"
  "os/signal"
  "syscall"
  "time"
)

var exit = make(chan string, 1)

func main() {
  go dealSignal()                       // 并发一个线程 dealSignal
  exited := make(chan struct{}, 1)      // 创建 channel 管道 exited，缓冲区为 1
  go channel1(exited)                   // 并发一个线程 channel1
  count := 0                            // 定义一个计数器
  t := time.Tick(time.Second )           // 定义一个时间t 

// 循环模块
Loop:                         
  for {
    select {                      // select 控制结构 ，随机执行，每个case必须是一个通讯操作，要么发送要么接收
    case <-t:                     // 接收 t
      count++                     
      fmt.Printf("main run %d\n", count)        // 打印 main run count
    case <-exited:                // 接收到 exited
      fmt.Println("main exit begin")            // 打印 main exit begim
      break Loop                                // 退出循环，进程成功退出
    }
  }
  fmt.Println("main exit end")                  // 打印 main exit end
}

func dealSignal() {
  c := make(chan os.Signal, 1)           // 创建管道 c，
  signal.Notify(c, os.Interrupt, syscall.SIGTERM)
  go func() {
    <-c
    exit <- "shutdown"                   // 发送 shutdown 到 exit
  }()
}

func channel1(exited chan<- struct{}) {
  t := time.Tick(time.Second * 5)
  count := 0
  for {
    select {
    case <-t:
      count++
      fmt.Printf("channel1 run %d\n", count)
    case <-exit:
      fmt.Println("channel1 exit")
      close(exited)
      return
    }
  }
}

这个例子首先并发出一个dealsign方法，用来接收关闭信号，如果接收到关闭信号后往exit channel发送一条消息，然后并发运行channel1，channel1中定了一个ticker，正常情况下channel1每秒打印第一个case语句，如果接收到exit的信号，进入第二个case，然后关闭传入的exited channel，那么main中的Loop，接收到exited关闭的信号后，打印“main exit begin”，然后退出循环，进程成功退出。这个例子演示了channel在goroutine中起到的传递消息的作用。这个例子是为了向大家展示channel在多个goroutine之间进行通信。

生产者/消费者案例

生产者goroutine负责将数据放入channel，消费者goroutine从channel中取出数据进行处理

package main

import (
    "fmt"
)

func main(){
    buf:=make(chan int)        // 创建通道 buf 、flg
    flg := make(chan int)
    go producer(buf)           // 创建线程
    go consumer(buf, flg)
    <- flg                     //等待接受完成
}

func producer(c chan int){
    defer close(c)             // 关闭channel
    for i := 0; i < 10; i++{
        c <- i                 // 阻塞，直到数据被消费者取走后，才能发送下一条数据
    }
}

func consumer(c, f chan int){
    for{
        if v, ok := <-c; ok{
            fmt.Print(v)        // 阻塞，直到生产者放入数据后继续读取数据
        }else{
            break
        }
    }
    f<-1 //发送数据，通知main函数已接受完成
}

运行结果如下