学习go语言编程之并发编程

发布时间 2023-08-14 22:57:32作者: nuccch

并发基础

并发包含如下几种主流的实现模型:

  • 多进程
  • 多线程
  • 基于回到的非阻塞/异步IO
  • 协程

协程

与传统的系统级线程和进程相比,协程最大的优势在于“轻量级”,可以轻松创建上百万个而不会导致系统资源枯竭,而线程和进程通常最多不超过1万个。
Golang在语言级别支持协程,叫goroutine

goroutine

goroutine是Golang中轻量级线程的实现,由Go运行时管理,使用go关键字来触发一个新的goroutine执行。
具体来说,在一个函数调用前加上关键字go,这次调用就会在一个新的goroutine中并发执行。
当被调用的函数返回时,这个goroutine也自动结束了。需要注意的是:如果这个函数有返回值,那么这个返回值会被丢弃。

func Add(a, b int) {
	z := a + b
	fmt.Println("z=", z)
}

func main() {
	for i := 0; i < 10; i++ {
		go Add(1, 1) // 在函数调用前使用关键字go,使得函数的调用是在goroutine中执行
	}
}

上述代码演示了如何在Golang中使用goroutine。

但是上述代码运行时并没有任何输出!原因:Go程序从初始化main package并执行main()函数开始,当main()函数返回时,程序退出,且程序并不会等待其他goroutine(非主goroutine)结束。

并发通信

在工程上,有2种最常见的并发通信模型:共享数据和消息。
被共享的数据可能有多种形式,如:内存数据块,磁盘文件,网络数据等。
如果是通过共享内存来实现并发通信,那就只能使用锁了。
Golang以并发编程作为语言的最核心优势,提供了另一种通信模型,即:以消息机制而非共享内存作为并发通信方式。
Golang提供的消息机制被称为channel。

channel

channel是Golang在语言级别提供的goroutine间通信方式,可以使用channel在两个或多个goroutine之间传递消息。
channel是进程内的通信方式,因此通过channel传递对象的过程和调用函数时的参数传递行为比较一致,比如也可以传递指针等。
channel是类型相关的,即:一个channel只能传递一种类型的值,这个类型需要在声明channel时指定。

基本语法

一般channel的声明形式为:

// 与声明一般变量的不同在于需要在类型前面加了关键字chan
// ElementType指定这个channel所能传递的元素类型
var chanName chan ElementType

示例:

// 声明一个传递类型为int的channel
var ch chan int

// 声明一个map,元素类型为bool的channel,即:这个channel传递的元素类型为map,map的值类型为bool
var m map[string] chan bool

定义一个channel也很简单,使用内置的函数make()即可:

// 声明并初始化了一个传递类型为int的channel
ch := make(chan int)

在channel的用法中,最常见的包括写入和读取。
将一个数据写入channel的语法:ch <- value,向channel写入数据通常会导致程序阻塞,直到有其他goroutine从这个channle中读取数据。
从channel中读取数据的语法是:value := <- ch,如果channel之前没有写入数据,那么从channel读取数据也会导致程序阻塞,直到channel中被写入数据为止。

select

Golang在语言级别支持select关键字,用于处理异步IO问题。
select与用法结构如下:

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

select的用法中,要求:每个case语句都必须是一个面向channel的操作。

如下是基于select的一段有趣的代码:

c := 0
ch := make(chan int, 1)
for {
    // 使用select随机向ch中写入0或1
    select {
    case ch <- 0:
    case ch <- 1:
    }

    i := <-ch
    fmt.Println("Received: ", i)

    c++
    if c > 10 {
        break
    }
}

缓冲机制

不带缓冲的channel,对于传递单个数据的场景可以接受,但是对于需要传递大量数据的场景就不合适了。
创建一个带缓冲的channel:

// 在调用make()时将缓冲区大小作为第二个参数传入即可
c := make(chan int, 1024)

带缓冲区的channel即使没有读取方,写入方也可以一直往channel中写入数据,在缓冲区填满之前都不会阻塞。

从带缓冲区的channel中读取数据可以使用与常规非缓冲channel完全一致的方法,但是也可以使用range关键字来实现更简便的循环读取。

// 使用range关键字来实现带缓冲区channel的循环读取
for v := range ch {
    fmt.Println("Received:", v)
}

超时机制

如果不能很好地处理超时问题,可能会导致goroutine永远阻塞而没有挽回的机会!
Golang中没有提供直接的超时处理机制,但是可以使用select很方便地解决超时问题(因为select的特点是只要其中一个case已经完成,程序就会继续往下执行,而不会考虑其他case的情况)。

ch := make(chan int, 1024)

// 首先,实现并执行一个匿名的超时等待函数
timeout := make(chan bool, 1)
go func() {
    time.Sleep(1e9) // 等待一秒钟
    timeout <- true
}()

// 然后,把timeout这个channel利用起来
select {
case <-ch:
    // 从目标channel中读取数据
case <-timeout:
    // 如果从目标channel中一直没有读取到数据,但是从timeout这个channel上读取到了数据
    // 这样就使用select机制可以避免永久等待的问题
    // 这是在Golang开发中避免channel通信超时的最有效办法
}

channel的传递

在Golang中channel本身也是一种原生类型,与map之类的类型地位一样,因此channel本身在定义后也可以通过channel来传递。
可以使用这个特性来实现管道,管道也是使用非常广泛的一种设计模式。

type PipeData struct {
	value   int
	handler func(int) int
	next    chan int
}

首先限定一个基本的数据结构PipeData,然后写一个常规的处理函数。只要定义一系列PipeData的数据结构并一起传递给这个函数,就可以达到流式处理数据的目的。

func handle(queue chan *PipeData) {
	for data := range queue {
		data.next <- data.handler(data.value)
	}
}

单向channel

单向channel只能用于发送或接收数据。
可以在将一个channel变量传递给一个函数时,通过指定其为单向channel变量,从而限制在该函数中可以对此channel执行的操作,比如只能往这个channel写,或者只能从这个channel读。

单向channel的声明非常简单,如下:

var ch1 chan int       // ch1是一个正常的channel,不是单向的
var ch2 chan<- float64 // ch2是一个用于只写float64数据单项channel
var ch3 <-chan int     // ch3是一个用于只读int数据的channel

单向channel的初始化:

ch4 := make(chan int)
ch5 := <-chan int(ch4) // ch5是一个单向读取的channel
ch6 := chan<- int(ch4) // ch6是一个单向写入的channel

如上,基于一个正常的channel可以实现单向channel的初始化。
即类型转换对于channel的意义:在单向channel和双向channel之间进行转换。

使用单向channel可以起到一种契约的作用:

func parse(ch <-chan int) {
	for value := range ch {
		fmt.Println("Received:", value)
	}
}

如上,除非这个函数的实现者使用了类型转换,否则这个函数就不会因为各种原因而对ch变量执行写操作,因而避免在ch中出现非期望的数据,从而很好地实践最小权限原则。

关闭channel

使用内置函数close()关闭channel。

close(ch)

如何判断一个channel是否已经关闭?可以通过在读取的时候使用多重返回值进行判断:

// 使用多重返回值检查channel是否已经关闭
val, ok := <-ch
if ok {
    // channel未关闭,可以正常使用返回值
    fmt.Println("Received:", val)
}

多核并行化

多核并行化是指尽量利用CPU多核特性来将任务并行化执行。
具体到Golang中,就是要知道CPU核心的数量,并针对性地将计算任务分解到多个goroutine中并行运行。

// 获取CPU核心数量
runtime.NumCPU()

出让时间片

使用runtime.Gosched()在每个goroutine中控制何时主动出让时间片给其他goroutine。

同步

同步锁

Golang的sync包中提供了两种锁类型:sync.Mutexsync.RWMutex
Mutex是最简单的锁类型,同时也比较暴力,当一个goroutine获得Mutex后,其他goroutine就只能等待这个goroutine释放该Mutex
RWMutex相对友好,是经典的单写多读模型。在读锁占用的情况下,会阻止写,但不阻止读。也就是多个goroutine可同时获取读锁,而写锁会阻止任何其他goroutine进来,整个锁相当于由该goroutine独占。获取读锁:sync.RWMutex.RLock(),获取写锁:sync.RWMutex.Lock()

对于这两种锁类型,任何一个Lock()RLock()均需要保证对应有Unlock()RUnlock()调用与之对应,否则可能导致等待该锁的所有goroutine处于饥饿状态,甚至可能导致死锁。
锁的典型使用模式如下:

// 先声明一个锁
var lock sync.Mutex
func foo() {
	lock.Lock()
	defer lock.Unlock() // defer关键字的方便之处
	// 获得锁之后需要执行的操作
}

全局唯一性操作

对于从全局的角度只需要运行一次的代码,比如全局初始化,Golang提供了一个Once类型来保证全局的唯一性操作。

var a string
var once sync.Once

func setup() {
	a = "Hello, World!"
	fmt.Println("初始化a")
}

func doPrint() {
	once.Do(setup) // 使用Once来控制函数在全局角度只会执行一次
	fmt.Println(a)
}

func twoPrint() {
	go doPrint()
	go doPrint()
}

如上示例代码,onceDo()方法可以保证在全局范围内只调用指定的函数一次,而且其他所有goroutine在调用到此语句时,将会先被阻塞,直到全局唯一的once.Do()调用结束之后才继续。

原子性操作

如果Golang中没有提供Once类型来保证全局唯一性操作,对于那些需要控制在全局只执行一次的操作来说,只能通过别的办法来处理了。

// 设置一个全局变量表示初始化操作是否完毕
var done bool = false

func setup() {
	a = "Hello, World!"
	done = true
	fmt.Println("初始化a")
}

func doPrint() {
	if !done {
		setup()
	}
	fmt.Println(a)
}

这段代码看起来合理,但是细看还是会有问题,因为setup()并不是一个原子性操作。这种写法可能会导致setup()被调用多次,从而无法达到全局只执行一次的目标。

为了更好地控制并行中的原子性操作,sync包中还包含了一个atomic子包,它提供了对于一些基础数据类型的原子操作函数。

// 比较和交换2个uint64类型数据
func CompareAndSwapInt64(addr *int64, old, new int64) (swapped bool)

有了这些原子操作函数,开发者就无需再为这样的操作专门添加Lock控制。

总结

关于Golang中并发编程有如下总结。

1.核心内容:协程
2.重要的关键字:changoselectdefer