Go并发编程

本文介绍 Go语言中的并发。

并发与并行

并行:指在同一时刻,有多条指令在多个处理器上同时执行。

并发:指在同一时刻只能有一条指令执行,但多个进程指令被快速的轮换执行,使得宏观上具有多个进程同时执行的效果,但在微观上并不是同时执行的,只是把时间分成若干段,使多个进程快速交替的执行。

Go语言并发优势

Go作为21世纪的C语言,从语言层面就支持了并发。

而且,并发程序的内存管理有时是非常复杂的,而Go语言提供了自动垃圾回收机制。

Go语言为并发编程而内置了上层API基于CSP(顺序通信进程)模型。

这就意味着显式锁都是可以避免的,因为Go语言通过安全的 channel通道 发送和接收数据以实现同步,这大大地简化了并发程序的编写。

一般情况下,一个普通的桌面计算机跑十几二十个线程就有点负载过大了,但是同样这台机器却可以轻松地让成百上千甚至过万个 goroutine 进行资源竞争。

goroutine

goroutine是什么?

goroutine 是Go并行设计的核心。

goroutine说到底其实就是协程,但是它比线程更小,十几个 goroutine 可能体现在底层就是五六个线程,Go语言内部帮你实现了 goroutine之间的内存共享。

执行 goroutine只需极少的栈内存(大概是4~5KB),当然会根据相应的数据伸缩。

也正因为如此,可同时运行成千上万个并发任务。

goroutine 比 thread更易用、更高效、更轻便。

创建goroutine

只需在函数调用语句前添加 go 关键字,就可以创建并发执行单元。

开发人员无需了解任何执行细节,调度器会自动将其安排到合适的系统线程上执行。

在并发编程里,我们通常想将一个过程切分成几块,然后让每个 goroutine各自负责一块工作。

当一个程序启动时,其主函数即在一个单独的 goroutine 中运行,我们叫它 main goroutine。

新的 goroutine会用 go 语句来创建。

实例:

package main

import (
   "fmt"
   "time"
)

func newTask() {
   for {
      fmt.Println("newTask")
      time.Sleep(time.Second)
   }
}

func main() {
   go newTask()

   for {
      fmt.Println("main")
      time.Sleep(time.Second)
   }
}

注意:主协程(main goroutine)退出,程序就结束了,子协程也就退出了

package main

import (
   "fmt"
   "time"
)

func NewTask() {
   for {
      fmt.Println("NewTask")
      time.Sleep(time.Second)
   }
}

func main() {

   go NewTask()

   i := 0
   for {
      i++
      fmt.Println("main i =",i)
      time.Sleep(time.Second)
      if i == 2 {
         break;
      }
   }
}

runtime包

runtime.Gosched()

Gosched() 用于让出 CPU时间片,让出当前 goroutine 的执行权限,调度器安排其他等待的任务运行,并在下次某个时候从该位置恢复执行。

package main

import (
   "fmt"
   "runtime"
)

func main() {

   go func() {
      for i := 0; i < 100; i++ {
         fmt.Println("go",i)
         //time.Sleep(time.Second)
      }
   }()

   for i := 0; i < 2; i++ {
      // 让出CPU执行的时间片,先让别的协程先执行一段时间,然后在重新回来继续执行
      runtime.Gosched()
      fmt.Println("hello")
   }
}

runtime.Goexit

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

package main

import (
   "fmt"
   "runtime"
)

func test() {
   defer fmt.Println("cccccccccccccccccccc")

   // 当使用 return时,是退出当前函数
   //return

   // 终止所在的协程
   runtime.Goexit()

   fmt.Println("dddddddddddddddddddd")
}

func main() {

   go func() {
      fmt.Println("aaaaaaaaaaaaaaaaaaaaa")

      test()

      fmt.Println("bbbbbbbbbbbbbbbbbbbbb")
   }()

   // 放一个死循环,防止程序退出
   for {

   }
}

runtime.GOMAXPROCS

调用 GOMAXPROCS() 用来设置可以并行计算的CPU核数的最大值,并返回之前的值。

package main

import (
   "fmt"
   "runtime"
)

func main() {
   n := runtime.GOMAXPROCS(1)
   fmt.Println(n)
}

注:在Go的最新版中,默认是使用CPU的全核心运行。

并发产生的问题:多任务资源竞争

并发情况下,对同一份数据进行操作的情况下,会出现资源竞争的情况。

示例1:对同一份数据进行修改

单协程(单进程单线程)的情况下,数据修改正常

package main

import (
   "fmt"
   "time"
)

var num = 100000000


func main() {
   func() {
      for i := 1; i <= 100000000/2; i++ {
         num -= 1
      }
   }()

   func() {
      for i := 1; i <= 100000000/2; i++ {
         num -= 1
      }
   }()

   for {
      time.Sleep(time.Second)
      fmt.Println(num)
   }
}

多协程的情况下,数据修改出现问题

package main

import (
   "fmt"
   "time"
)

var num = 100000000


func main() {
   go func() {
      for i := 1; i <= 100000000/2; i++ {
         num -= 1
      }
   }()

   go func() {
      for i := 1; i <= 100000000/2; i++ {
         num -= 1
      }
   }()

   for {
      time.Sleep(time.Second)
      fmt.Println(num)
   }
}

使用其他语言中的解决方案,对操作同一份数据的代码加锁

package main

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

var num = 100000000

var lock sync.Mutex

func main() {
   go func() {
      lock.Lock()
      for i := 1; i <= 100000000/2; i++ {
         num -= 1
      }
      lock.Unlock()
   }()

   go func() {
      lock.Lock()
      for i := 1; i <= 100000000/2; i++ {
         num -= 1
      }
      lock.Unlock()
   }()

   for {
      time.Sleep(time.Second)
      fmt.Println(num)
   }
}

示例2:对同一资源使用

多协程的情况下,数据是乱序的

package main

import (
   "fmt"
   "time"
)

// 定义一个公共资源 - 打印机
func Printer(s string) {
   for _, data := range s {
      fmt.Printf("%c", data)
      time.Sleep(time.Second)
   }
}

// 定义一个使用打印机的人 person1
func person1() {
   Printer("hello")
}

// 定义一个使用打印机的人 person2
func person2() {
   Printer("world")
}

func main() {

   go person1()

   go person2()

   // 使用死循环,不让主程序退出
   for {

   }
}

使用锁解决乱序问题

package main

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

var lock sync.Mutex

// 定义一个公共资源 - 打印机
func Printer(s string) {
   lock.Lock()
   for _, data := range s {
      fmt.Printf("%c", data)
      time.Sleep(time.Second)
   }
   lock.Unlock()
}

// 定义一个使用打印机的人 person1
func person1() {
   Printer("hello")
}

// 定义一个使用打印机的人 person2
func person2() {
   Printer("world")
}

func main() {

   go person1()

   go person2()

   // 使用死循环,不让主程序退出
   for {

   }
}

注:Go语言中解决并发问题使用通道channel

通道 channel

goroutine运行在相同的地址空间,因此访问共享内存必须做好同步。

goroutine奉行通过通信来共享内存,而不是共享内存来通信。

引用类型 channel 是 CSP(顺序通信进程)模式的具体实现,用于多个 goroutine通讯。其内部实现了同步,确保并发安全。

channel 类型 和 map 类似,channel也对应一个make创建的底层数据结构的引用。

当我们复制一个 channel或用于函数参数传递时,我们只是拷贝了一个 channel引用,因此调用者和被调用者将引用同一个 channel对象。和其它的引用类型一样,channel的零值也是 nil。

使用channel

定义一个 channel,需要定义发送到 channel的值的类型。

channel可以使用内置的 make()函数来创建。

make(chan Type)    // 等价于make(chan Type, 0)

make(chan Type, capacity)

channel 通过操作符 <- 来接收和发送数据

channel <- value    // 发送 value 到 channel

<-channel        // 接收并将其丢弃

x := <-channel    // 从 channel中接收数据,并赋值给x

x, ok := <-channel    // 功能同上,同时检查通道是否已关闭或者是否为空

示例1:对同一份数据进行修改

package main

import (
   "fmt"
   "time"
)

var num = 100000000

func main() {

   ch := make(chan int)

   go func() {
      for i := 1; i <= 100000000/2; i++ {
         num -= 1
      }
      ch <- 666  // 上面执行完后,在往channel中写入数据
   }()

   go func() {
      <-ch // 读取channel数据,如果没有数据就阻塞在这里,等待channel有数据并读取后,在往后执行
      for i := 1; i <= 100000000/2; i++ {
         num -= 1
      }
   }()

   for {
      time.Sleep(time.Second)
      fmt.Println(num)
   }
}

示例2:对同一资源使用

package main

import (
   "fmt"
   "time"
)

var ch = make(chan int)

func Printer(s string) {
   for _, data := range s {
      fmt.Printf("%c", data)
      // goroutine 遇到阻塞任务就会让出CPU的时间片(执行权),让其他goroutine先执行
      time.Sleep(time.Second)
   }
}

func Person1() {
   Printer("hello")
   ch <- 666  // 上面执行完后,在往channel中写入数据
}

func Person2() {
   <-ch      // 读取channel数据,如果没有数据就阻塞在这里,等待channel有数据并读取后,在往后执行
   Printer("world")
}

func main() {
   go Person1()

   go Person2()

   // 使用死循环防止主程序退出
   for {

   }
}

无缓冲的channel

无缓冲的通道是指在接收前没有能力保存任何值的通道。

这种类型的通道要求发送 goroutine和接收 goroutine 同时准备好,才能完成发送和接收操作。

如果两个 goroutine没有同时准备好,通道会导致先执行发送或接收操作的goroutine阻塞等待。

这种对通道进行发送和接收的交互行为本身就是同步的。

其中任意一个操作都无法离开另一个操作单独存在。

ch := make(chan Type)    // 等价于 make(chan Type, 0)

有缓冲的channel

有缓冲的通道是一种在接收前能存储一个或者多个值的通道。

这种类型的通道并不强制要求 goroutine 之间必须同时完成发送和接收。

通道会阻塞发送和接收动作的条件也不同。

只有在通道中没有要接收的值时,接收动作才会阻塞。

只有在通道没有可用缓冲区容纳被发送的值时,发送动作才会阻塞。

ch := make(chan Type, capacity)

有缓冲区的通道和无缓冲区的通道之间的一个很大的不同:

无缓冲的通道保证进行发送和接收的 goroutine 会在同一时间进行数据交换;有缓冲的通道没有这种保证。

如果给定了一个缓冲区容量,通道就是异步的。

只要缓冲区有未使用空间用于发送数据,或还包含可以接收的数据,那么其通信就会无阻塞的进行。

关闭channel

如果发送者知道,没有更多的值需要发送到 channel 的话,那么让接收者也能及时知道没有多余的值可接收将是有用的,因为接收者可以停止不必要的接收等待。

这可以通过内置的 close 函数来关闭 channel 实现。

package main

import "fmt"

func main() {
   ch := make(chan int)

   go func() {
      for i := 0; i < 5; i++ {
         ch <- i
      }
      // 闭关channel
      close(ch)
   }()

   for {
      // ok为true,说明channel没有关闭,为false说明管道已经关闭
      if data, ok := <-ch; ok {
         fmt.Println(data)
      } else {
         break
      }
   }
}

注意点:

  • channel不像文件一样需要经常去关闭,只有当你确实没有任何发送数据了,或者你想显式的结束 range循环之类的,才去关闭 channel。
  • 关闭channel后,无法向 channel再发送数据(引发 panic错误后导致接收立即返回零值)。
  • 关闭 channel后,可以继续向 channel接收数据。
  • 对于 nil channel,无论收发都会被阻塞。

range

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

package main

import "fmt"

func main() {
   ch := make(chan int)

   go func() {
      for i := 0; i < 5; i++ {
         ch <- i
      }
      // 闭关channel
      close(ch)
   }()

   for num := range ch {
      fmt.Println(num)
   }

   //for {
   // // ok为true,说明channel没有关闭,为false说明管道已经关闭
   // if data, ok := <-ch; ok {
   //    fmt.Println(data)
   // } else {
   //    break
   // }
   //}
}

单方向的channel

var ch1 chan int
var ch2 chan <- float64
var ch3 <-chan int

默认情况下,通道是双向的,也就是,既可以往里面发送数据也可以从里面接收数据。

但是,我们经常见一个通道作为参数进行传递而只希望对方是单向使用的,要么只让它发送数据,要么只让它接收数据,这时候我们可以指定通道的方向。

双向通道可以向单向通道转换,单向无法转换为双向

// 创建一个channel,双向的
ch := make(chan int)

// 双向channel能隐式转换为单向channel
var writeCh chan <- int = ch    // 只能写不能读

writeCh <- 666  // 写

// Error
// <- writeCh    // 读


var readCh <-chan int = ch    // 只能读不能写

<-readCh       // 读

// Error
// readCh <- 666 // 写

// 单向无法转换为双向
// var ch2 chan int = writeCh

实例

package main

import "fmt"

func producer(out chan <- int) {
   for i :=0; i < 10; i++ {
      out <- i * i
   }
   close(out)
}

func consumer(in <-chan int) {
   for num := range in {
      fmt.Println("num =", num)
   }
}

func main() {

   ch := make(chan int)

   go producer(ch)

   consumer(ch)

}

定时器

1、Timer

Timer是一个定时器,代表未来的一个单一事件,你可以告诉 timer你要等待多长时间,它提供一个channel,在将来的那个时间那个 channel提供了一个时间值。

实例:延时任务

// 延时2s后打印一句话
timer := time.NewTimer(time.Second * 2)
<- timer.C
fmt.Println("时间到")

// 延时2s后打印一句话
<-time.After(time.Second * 2)
fmt.Println("时间到")

// 延时2s后打印一句话
time.Sleep(time.Second * 2)
fmt.Println("时间到")

停止定时器

package main

import (
   "fmt"
   "time"
)

func main() {

   timer := time.NewTimer(time.Second * 3)

   go func() {
      <-timer.C
      fmt.Println("时间到,子协程开始打印!")
   }()

   timer.Stop()   // 停止定时器

   // 防止主协程退出
   for {

   }
}

重置定时器

package main

import (
   "fmt"
   "time"
)

func main() {

   timer := time.NewTimer(time.Second * 5)

   go func() {
      <-timer.C
      fmt.Println("时间到,子协程开始打印!")
   }()

   timer.Reset(time.Second)   // 重置定时器为1秒

   // 防止主协程退出
   for {

   }
}

2、Ticker

Ticker是一个定时触发的计时器,它会以一个间隔(interval)往channel发送一个事件(当前时间),而channel的接收者可以以固定的时间间隔从channel中读取事件。

实例:定时任务

package main

import (
   "fmt"
   "time"
)

func main() {

   ticker := time.NewTicker(time.Second * 1)

   i := 0
   for {
      <-ticker.C
      i++
      fmt.Println(i)

      // 退出条件
      if i == 5 {
         ticker.Stop()     // 停止定时器
         break
      }
   }
}

select

Go里面提供了一个关键字 select,通过select可以监听channel上的数据流动。

select的用法与 switch语法非常类似,由select开始一个新的选择块,每个选择条件由 case 语句来描述。

与switch语句可以选择任何可使用相等比较的条件相比,select有比较多的限制,其中最大的一条限制就是每个case语句里必须是一个IO操作。

在一个select语句中,Go语言会按顺序从头到尾评估每一个发送和接收的语句。

select {
    case <-chan1:
        // 如果 chan1成功读到数据,则进行该 case处理语句
    case chan2 <- 1:
        // 如果成功向 chan2写入数据,则进行该 case处理语句
    default:
        // 如果上面都没有成功,则进入default处理流程
}

如果其中任意一条语句可以继续执行(即没有阻塞),那么就从那些可以执行的语句中任意选择一条来使用。

如果没有任意一条语句可以执行(即所有的通道都被阻塞),那么有两种可能的情况:

  • 如果给出了default语句,那么就会执行default语句,同时程序的执行会从select语句中恢复。
  • 如果没有default语句,那么 select语句将被阻塞,直到至少有一个通信可以进行下去。

实例:通过select实现斐波那契数列

package main

import "fmt"

func fibonacci(ch chan<-int, quit <-chan bool) {
   n, m := 1, 1
   for {
      select {
      case ch <- n:
         n, m = m, m + n
      case flag := <-quit:
         fmt.Println("flag = ", flag)
         return
      }
   }
}

func main() {
   ch := make(chan int)
   quit := make(chan bool)

   go func() {
      for i := 0; i < 8; i++ {
         num := <-ch
         fmt.Println(num)
      }
      quit <- true
   }()

   fibonacci(ch, quit)
}

超时

有时候会出现 goroutine阻塞的情况,那么我们如何避免整个程序进入阻塞的情况呢?

我们可以利用 select来设置超时

package main

import (
   "fmt"
   "time"
)

func main() {
   ch := make(chan int)
   quit := make(chan bool)

   go func() {
      for {
         select {
         case num := <-ch:
            fmt.Println("num = ", num)
         case <-time.After(3 * time.Second):
            fmt.Println("超时")
            quit <- true
         }
      }
   }()

   <-quit
   fmt.Println("程序结束")
}

相关文章

感觉本篇文章不错,对你有收获?

¥我要小额赞助,鼓励作者写出更好的教程
80 160 120

作者:

  • 出处: https://www.mi360.cn/articles/238
  • 本文版权归作者,欢迎转载,但未经作者同意必须保留 此段声明,且在文章页面明显位置给出原文连接,否则保留追究法律责任的权利。

评论区

最新评论

扫码关注