Go语言基础（十二）：并发编程

文章目录

一、前言
二、goroutine
- 1、使用goroutine
- 2、启动goroutine示例
- 3、main优雅谢幕（sync.WaitGroup）
三、goroutine与线程
- 1、可增长的栈
- 2、goroutine调度
- 3、GOMAXPROCS
四、channel
- 1、channel类型
- 2、创建channel
- 3、channel操作
- - （1）发送
  - （2）接收
  - （3）关闭
- 4、无缓冲的通道
- 5、有缓冲的通道
- 6、for range从通道循环取值
- 7、单向通道
- 8、通道总结
五、worker pool（goroutine池）
六、select多路复用
七、上下文Context
- 1、如何停止一个子协程（channel实现）
- 2、基于 Context 停止子协程
八、并发安全与锁
- 1、互斥锁
- 2、读写互斥锁
- 3、sync.Once
- - （1）并发控制---执行一次（Java双重检验锁）
  - （2）并发安全的单例模式
- 4、sync.Map
九、生产者消费者（kubernetes队列）
十、原子操作
- 1、atomic包
- 2、示例（比较互斥锁、原子操作的性能）

一、前言

Go语言的并发通过goroutine实现。goroutine类似于线程，属于用户态的线程，我们可以根据需要创建成千上万个goroutine并发工作。goroutine是由Go语言的运行时（runtime）调度完成，而线程是由操作系统调度完成。

Go语言还提供channel在多个goroutine间进行通信。goroutine和channel是 Go 语言秉承的 CSP（Communicating Sequential Process）并发模式的重要实现基础。

由于其是用户态线程，没有从用户态到核心态的切换的开销，因此goroutine是非常轻量级的线程。其实goroutine和channel之间的关系，相当于进程与队列之间的关系。

Go语言中的goroutine就是这样一种机制，goroutine的概念类似于线程，但 goroutine是由Go的运行时（runtime）调度和管理的。Go程序会智能地将 goroutine 中的任务合理地分配给每个CPU。Go语言之所以被称为现代化的编程语言，就是因为它在语言层面已经内置了调度和上下文切换的机制。

二、goroutine

在Go语言编程中你不需要去自己写进程、线程、协程，你的技能包里只有一个技能–goroutine，当你需要让某个任务并发执行的时候，你只需要把这个任务包装成一个函数，开启一个goroutine去执行这个函数就可以了，就是这么简单粗暴。

1、使用goroutine

Go语言中使用goroutine非常简单，只需要在调用函数的时候在前面加上go关键字，就可以为一个函数创建一个goroutine。

一个goroutine必定对应一个函数，可以创建多个goroutine去执行相同的函数。

一个函数，一般定义成一个要做的任务。

2、启动goroutine示例

func hello() {fmt.Println("Hello Goroutine!")
}
func main() {hello() //这是我们一般执行代码的逻辑fmt.Println("main goroutine done!")
}

接下来我们需要，用goroutine启动一个线程去完成hello任务：

func main() {go hello() // 加go关键字，启动另外一个goroutine线程去执行hello函数fmt.Println("main goroutine done!")
}

如果执行了上述代码你就会发现，执行结果只打印了main goroutine done!，并没有打印Hello Goroutine!。为什么呢？
答： Go开启一个线程去执行hello任务之后，然后直接就print输出main goroutine done!，主进程就结束了，然后所属它的子线程也会被杀死。因为创建线程有时间开销，代码执行速度是非常快的。所以子线程还没来得及打印Hello Goroutine!就被杀死了。

所以我们要想办法让main函数等一等hello函数，最简单粗暴的方式就是time.Sleep了。

func main() {go hello() // 启动另外一个goroutine去执行hello函数fmt.Println("main goroutine done!")time.Sleep(time.Second)
}

执行上面的代码你会发现，这一次先打印main goroutine done!，然后紧接着打印Hello Goroutine!。

3、main优雅谢幕（sync.WaitGroup）

sync.WaitGroup方法如下：

上面的sleep去等子线程结束之后，主进程再退出，存在可能等多了，可能等少了这种情况，因此需要有一种优雅的方式去结束主进程（因此下面使用了 sync.WaitGroup 实现main优雅谢幕）。

var wg sync.WaitGroup //声明一个变量wgfunc hello(i int) {defer wg.Done() // goroutine结束，wg-=1fmt.Println("Hello Goroutine!", i)
}
func main() {for i := 0; i < 10; i++ {wg.Add(1) // 启动一个goroutine，wg+=1go hello(i)}wg.Wait() // 当监听到wg为0（即所有goroutine结束了），则执行后面的逻辑
}

除此之外，上面还利用for循环，生成了多个goroutine。值得注意的是:上面的10个goroutine只要有一个线程发生了阻塞，这个主进程都不会退出，这样会浪费大量资源，一定要确保goroutine不会发生阻塞。

三、goroutine与线程

1、可增长的栈

OS线程栈内存（操作系统线程）一般都是固定的（通常为2MB）,一个goroutine栈内存在其生命周期开始时只有很小的栈（典型情况下2KB），goroutine的栈内存不是固定的，他可以按需增大和缩小，goroutine的栈大小限制可以达到1GB，虽然极少会用到这么大。所以在Go语言中一次创建十万左右的goroutine也是可以的。

2、goroutine调度

GPM是Go语言运行时（runtime）层面的实现，是go语言自己实现的一套调度系统。区别于操作系统调度OS线程。

G很好理解，就是个goroutine的，里面除了存放本goroutine信息外还有与所在P的绑定等信息。
P管理着一组goroutine队列，P里面会存储当前goroutine运行的上下文环境（函数指针，堆栈地址及地址边界），P会对自己管理的goroutine队列做一些调度（比如把占用CPU时间较长的goroutine暂停、运行后续的goroutine等等）当自己的队列消费完了就去全局队列里取，如果全局队列里也消费完了会去其他P的队列里抢任务。
M（machine）是Go运行时（runtime）对操作系统内核线程的虚拟， M与内核线程一般是一一映射的关系，一个groutine最终是要放到M上执行的，这个数量一般设置为cpu核心数；

P与M一般也是一一对应的。他们关系是： P管理着一组G挂载在M上运行。当一个G长久阻塞在一个M上时，runtime会新建一个M，阻塞G所在的P会把其他的G 挂载在新建的M上。当旧的G阻塞完成或者认为其已经死掉时回收旧的M。

P的个数是通过runtime.GOMAXPROCS设定（最大256），Go1.5版本之后默认为物理线程数，P和M数量一般一致。在并发量大的时候会增加一些P和M，但不会太多，切换太频繁的话得不偿失。

上面说到P的个数默认等于CPU核数，每个M必须持有一个P才可以执行G，一般情况下M的个数会略大于P的个数，这多出来的M将会在G产生系统调用时发挥作用。类似线程池，Go也提供一个M的池子，需要时从池子中获取，用完放回池子，不够用时就再创建一个。

M的数量：

go 语言本身的限制：go 程序启动时，会设置 M 的最大数量，默认 10000. 但是内核很难支持这么多的线程数，所以这个限制可以忽略。
runtime/debug 中的 SetMaxThreads 函数，设置 M 的最大数量一个 M 阻塞了，会创建新的 M。

单从线程调度讲，Go语言相比起其他语言的优势在于OS线程是由OS内核来调度的，goroutine则是由Go运行时（runtime）自己的调度器调度的，这个调度器使用一个称为m:n调度的技术（复用/调度m个goroutine到n个OS线程）。其一大特点是goroutine的调度是在用户态下完成的， 不涉及内核态与用户态之间的频繁切换，包括内存的分配与释放，都是在用户态维护着一块大的内存池， 不直接调用系统的malloc函数（除非内存池需要改变），成本比调度OS线程低很多。另一方面充分利用了多核的硬件资源，近似的把若干goroutine均分在物理线程上，再加上本身goroutine的超轻量，以上种种保证了go调度方面的性能。

点我了解更多哦

进程线程协程核心区别：

1、进程之间切换的开销，有刷新TLB、重新生成虚拟地址，页表，一个进程的创建前期会有很多缺页异常，因此效率非常慢2、线程之间切换的开销，线程和进程其实都是一个task struct，线程又叫轻量级进程，本质上其实就是进程，只不过共享了内存地址、资源，因此它切换可以复用虚拟地址、页表、TLB快表，所以它比进程切换开销小，但线程的切换还是要依靠系统调用，要陷入到内核态，即用户态到内核态的切换。3、go协程之间切换的开销，连线程的用户态、内核态切换都没有了，因为它的切换是通过GMP中的P调度器在用户态维护M和G之间的绑定。而且一般一个cpu分给M，其M的P便会一直使用这个cpu去一个个处理G,直到发生cpu时间片轮转。

3、GOMAXPROCS

Go运行时的调度器使用GOMAXPROCS参数来确定需要使用多少个OS线程来同时执行Go代码。默认值是机器上的CPU核心数。例如在一个8核心的机器上，调度器会把Go代码同时调度到8个OS线程上（GOMAXPROCS是m:n调度中的n）。

Go语言中可以通过runtime.GOMAXPROCS()函数设置当前程序并发时占用的CPU逻辑核心数。

Go1.5版本之前，默认使用的是单核心执行。Go1.5版本之后，默认使用全部的CPU逻辑核心数。

我们可以通过将任务分配到不同的CPU逻辑核心上实现并行的效果，这里举个例子：

func main() {runtime.GOMAXPROCS(1)go a()go b()time.Sleep(time.Second)
}

两个任务只有一个逻辑核心，此时是做完一个任务再做另一个任务。将逻辑核心数设为2，此时两个任务并行执行，代码如下。

func main() {runtime.GOMAXPROCS(2)go a()go b()time.Sleep(time.Second)
}

Go语言中的操作系统线程和goroutine的关系：

一个操作系统线程对应用户态多个goroutine。
go程序可以同时使用多个操作系统线程。
goroutine和OS线程是多对多的关系，即m:n。

四、channel

单纯地将函数并发执行是没有意义的。函数与函数间需要交换数据才能体现并发执行函数的意义。

虽然可以使用共享内存进行数据交换，但是共享内存在不同的goroutine中容易发生竞态问题。为了保证数据交换的正确性，必须使用互斥量对内存进行加锁，这种做法势必造成性能问题。

Go语言的并发模型是CSP（Communicating Sequential Processes），提倡通过通信共享内存而不是通过共享内存而实现通信。

如果说goroutine是Go程序并发的执行体，channel就是它们之间的连接。channel是可以让一个goroutine发送特定值到另一个goroutine的通信机制。goroutine之间通信通过channel，类似进程之间通信通过队列

Go 语言中的通道（channel）是一种特殊的类型。通道像一个传送带或者队列，总是遵循先入先出（First In First Out）的规则（队列？很像！），保证收发数据的顺序。每一个通道都是一个具体类型的导管，也就是声明channel的时候需要为其指定元素类型。

1、channel类型

channel是一种类型，一种引用类型。声明通道类型的格式如下：

var 变量 chan 元素类型

举栗：

var ch1 chan int   // 声明一个传递整型的通道
var ch2 chan bool  // 声明一个传递布尔型的通道
var ch3 chan []int // 声明一个传递int切片的通道

2、创建channel

通道是引用类型，通道类型的空值是nil。

var ch chan int
fmt.Println(ch) // <nil>

声明的通道后需要使用make函数初始化之后才能使用。

创建channel的格式如下：

make(chan 元素类型, [缓冲大小])

举栗：

ch4 := make(chan int)
ch5 := make(chan bool)
ch6 := make(chan []int)

3、channel操作

通道有发送（send）、接收(receive）和关闭（close）三种操作。
通道定义如下：

ch := make(chan int)

（1）发送

将一个值发送到通道中。

ch <- 10 // 把10发送到ch中

（2）接收

从一个通道中接收值。

x := <- ch // 从ch中接收值并赋值给变量x
<-ch       // 从ch中接收值，忽略结果

（3）关闭

我们通过调用内置的close函数来关闭通道。

close(ch)

关于关闭通道需要注意的事情是，只有在通知接收方goroutine所有的数据都发送完毕的时候才需要关闭通道。通道是可以被垃圾回收机制回收的，它和关闭文件是不一样的，在结束操作之后关闭文件是必须要做的，但关闭通道不是必须的。

关闭后的通道有以下特点：

对一个关闭的通道再发送值就会导致panic。
对一个关闭的通道进行接收会一直获取值直到通道为空。
对一个关闭的并且没有值的通道执行接收操作会得到对应类型的零值。
关闭一个已经关闭的通道会导致panic。

4、无缓冲的通道

无缓冲的通道又称为阻塞的通道。我们来看一下下面的代码：

func main() {ch := make(chan int)ch <- 10fmt.Println("发送成功")
}

上面这段代码能够通过编译，但是执行的时候会出现以下错误：

fatal error: all goroutines are asleep - deadlock!goroutine 1 [chan send]:
main.main().../src/github.com/Q1mi/studygo/day06/channel02/main.go:8 +0x54

为什么会出现deadlock错误呢？

因为我们使用ch := make(chan int)创建的是无缓冲的通道，无缓冲的通道只有在有人接收值的时候才能发送值。就像你住的小区没有快递柜和代收点，快递员给你打电话必须要把这个物品送到你的手中，简单来说就是无缓冲的通道必须有接收才能发送。

上面的代码会阻塞在ch <- 10这一行代码形成死锁，那如何解决这个问题呢？

一种解决方法是启用一个goroutine去接收值，例如：

func recv(c chan int) {ret := <-cfmt.Println("接收成功", ret)
}
func main() {ch := make(chan int)go recv(ch) // 启用goroutine从通道接收值ch <- 10fmt.Println("发送成功")
}

无缓冲通道上的发送操作会阻塞，直到另一个goroutine在该通道上执行接收操作，这时值才能发送成功，两个goroutine将继续执行。相反，如果接收操作先执行，接收方的goroutine将阻塞，直到另一个goroutine在该通道上发送一个值。

使用无缓冲通道进行通信将导致发送和接收的goroutine同步化。因此，无缓冲通道也被称为同步通道。

5、有缓冲的通道

解决上面问题的方法还有一种就是使用有缓冲区的通道。我们可以在使用make函数初始化通道的时候为其指定通道的容量，例如：

func main() {ch := make(chan int, 1) // 创建一个容量为1的有缓冲区通道ch <- 10fmt.Println("发送成功")
}

只要通道的容量大于零，那么该通道就是有缓冲的通道，通道的容量表示通道中能存放元素的数量。就像你小区的快递柜只有那么个多格子，格子满了就装不下了，就阻塞了，等到别人取走一个快递员就能往里面放一个。

我们可以使用内置的len函数获取通道内元素的数量，使用cap函数获取通道的容量，虽然我们很少会这么做。

6、for range从通道循环取值

当向通道中发送完数据时，我们可以通过close函数来关闭通道。
当通道被关闭时，再往该通道发送值会引发panic，从该通道取值会先取完通道中的值，再然后取到一直为零值。那如何判断通道是否被关闭了呢？

// channel 练习
func main() {ch1 := make(chan int)ch2 := make(chan int)// 开启goroutine将0~100的数发送到ch1中go func() {for i := 0; i < 100; i++ {ch1 <- i}close(ch1)}()// 开启goroutine从ch1中接收值，并将该值的平方发送到ch2中go func() {for {i, ok := <-ch1 // 第一种：通道关闭后再取值ok=falseif !ok {break}ch2 <- i * i}close(ch2)}()// 在主goroutine中从ch2中接收值打印for i := range ch2 { // 第二种：通道关闭后会退出for range循环fmt.Println(i)}
}

7、单向通道

有的时候我们会将通道作为参数在多个任务函数间传递，很多时候我们在不同的任务函数中使用通道都会对其进行限制，比如限制通道在函数中只能发送或只能接收。

Go语言中提供了单向通道来处理这种情况。如下：

func counter(in chan<- int) {for i := 0; i < 100; i++ {in <- i}close(in)
}func squarer(out chan<- int, in <-chan int) {for i := range in {out <- i * i}close(out)
}
func printer(out <-chan int) {for i := range out {fmt.Println(i)}
}func main() {ch1 := make(chan int)ch2 := make(chan int)go counter(ch1)go squarer(ch2, ch1)printer(ch2)
}

其中，

chan<- int是一个只写单向通道（发送进）（只能对其写入int类型值），可以对其执行发送操作但是不能执行接收操作；
<-chan int是一个只读单向通道（接收）（只能从其读取int类型值），可以对其执行接收操作但是不能执行发送操作。

在函数传参及任何赋值操作中可以将双向通道转换为单向通道（类似自动类型转换），但反过来是不可以的。

8、通道总结

channel常见的异常总结，如下图：

关闭已经关闭的channel也会引发panic。

五、worker pool（goroutine池）

在工作中我们通常会使用可以指定启动的goroutine数量–worker pool模式，控制goroutine的数量，防止goroutine泄漏和暴涨。

一个简易的work pool示例代码如下：

func worker(id int, jobs <-chan int, results chan<- int) {for j := range jobs { // 当某个goroutine先做完一个任务时，这个for循环让这个goroutine又去通道接收任务参数，进行下一个任务fmt.Printf("worker:%d start job:%d\n", id, j)time.Sleep(time.Second)fmt.Printf("worker:%d end job:%d\n", id, j)results <- j * 2}
}func main() {jobs := make(chan int, 100)results := make(chan int, 100)// 开启了一个包含3个goroutine的池子// 池子等于两个for循环+一个channel，外部循环控制池子大小，内部循环保证线程一直在寻找新的任务做，channel是任务发布中心for w := 1; w <= 3; w++ {go worker(w, jobs, results)}// 5个任务for j := 1; j <= 5; j++ {jobs <- j}close(jobs)// 输出结果for a := 1; a <= 5; a++ {<-results}
}

六、select多路复用

在某些场景下我们需要同时从多个通道接收数据。通道在接收数据时，如果没有数据可以接收将会发生阻塞。你也许会写出如下代码使用遍历的方式来实现：

for{// 尝试从ch1接收值data, ok := <-ch1// 尝试从ch2接收值data, ok := <-ch2…
}

这种方式虽然可以实现从多个通道接收值的需求，但是运行性能会差很多。为了应对这种场景，Go内置了select关键字，可以同时响应多个通道的操作。

select{case <-ch1:...case data := <-ch2:...case ch3<-data:...default:默认操作
}

举个小例子来演示下select的使用：

func main() {ch := make(chan int, 1)for i := 0; i < 10; i++ {select {case x := <-ch: // 有东西能取出就取出打印fmt.Println(x)case ch <- i: // 没满能写就写}}
}

使用select语句能提高代码的可读性。

可处理一个或多个channel的发送/接收操作。
如果多个case同时满足，select会随机选择一个。
对于没有case的select{}会一直等待，可用于阻塞main函数。

七、上下文Context

使用场景：传递变量、超时、取消操作或者一些异常情况，往往需要进行抢占操作或者中断后续操作

Context 是设置截止日期、同步信号，传递请求相关值的结构体

type Context interface {Deadline() (deadline time.Time, ok bool)Done() <-chan struct{}Err() errorValue(key interface{}) interface{}
}

• 用法

context.Background
context.TODO
context.WithDeadline
context.WithValue
context.WithCancel

1、如何停止一个子协程（channel实现）

done := make(chan bool)
go func() {for {select {case <-done:fmt.Println("done channel is triggerred, exit child go routine")return} }
}()
close(done)

2、基于 Context 停止子协程

package mainimport ("context""fmt""time"
)func main() {baseCtx := context.Background()ctx := context.WithValue(baseCtx, "a", "b")// 可以通过context为子协程传一些父进程的变量，比如上面就把a=b传到c context.go func(c context.Context) {fmt.Println(c.Value("a"))}(ctx)// 生成了一个timeout context，再过time.Second时，context就会取消.timeoutCtx, cancel := context.WithTimeout(baseCtx, time.Second)res := make(chan int, 1)defer cancel()go func(ctx context.Context, out chan<- Value) error {for {v, err := DoSomething(ctx)if err != nil {return err}select {// DoSomething如果没在规定的timeout时间内完成就直接报错退出.case <-ctx.Done():return fmt.Errorf("child process interrupt")// 这个分支先拿到值，那么就说明在规定时间内DoSomething，正常.case out <- v:fmt.Printf("This task is finish")}}}(timeoutCtx, res)select {case <-timeoutCtx.Done():time.Sleep(1 * time.Second)fmt.Println("main process exit!")}// time.Sleep(time.Second * 5)
}

八、并发安全与锁

有时候在Go代码中可能会存在多个goroutine同时操作一个资源（临界区），这种情况会发生竞态问题（数据竞态）。类比现实生活中的例子有十字路口被各个方向的的汽车竞争；还有火车上的卫生间被车厢里的人竞争（资源竞态）。

举个例子：

var x int64
var wg sync.WaitGroupfunc add() {for i := 0; i < 100000; i++ {x = x + 1}wg.Done()
}
func main() {wg.Add(2)go add()go add()wg.Wait()fmt.Println(x)
}

上面的代码中我们开启了两个goroutine去累加变量x的值，这两个goroutine在访问和修改x变量的时候就会存在数据竞争，导致最后的结果与期待的不符。

1、互斥锁

互斥锁是一种常用的解决数据竞态的方法，它能够保证同时只有一个goroutine可以访问共享资源。Go语言中使用sync包的Mutex类型来实现互斥锁。使用互斥锁来修复上面代码的问题：

var x int64
var wg sync.WaitGroup
var lock sync.Mutexfunc add() {for i := 0; i < 5000; i++ {lock.Lock() // 加锁x = x + 1lock.Unlock() // 解锁}wg.Done()
}
func main() {wg.Add(2)go add()go add()wg.Wait()fmt.Println(x)
}

使用互斥锁能够保证同一时间有且只有一个goroutine进入临界区，其他的goroutine则在等待锁；当互斥锁释放后，等待的goroutine才可以获取锁进入临界区，多个goroutine同时等待一个锁时，唤醒的策略是随机的。

2、读写互斥锁

互斥锁是完全互斥的，但是有很多实际的场景下是读多写少（读写锁应用场景）的，当我们并发的去读取一个资源不涉及资源修改的时候是没有必要加锁的，这种场景下使用读写锁是更好的一种选择。读写锁在Go语言中使用sync包中的RWMutex类型。

读写锁分为两种：读锁和写锁。当一个goroutine获取读锁之后，其他的goroutine如果是获取读锁会继续获得锁，如果是获取写锁就会等待；当一个goroutine获取写锁之后，其他的goroutine无论是获取读锁还是写锁都会等待（共享读、阻塞写）。

读写锁示例：

var (x      int64wg     sync.WaitGrouplock   sync.Mutexrwlock sync.RWMutex
)func write() {// lock.Lock()   // 加互斥锁rwlock.Lock() // 加写锁x = x + 1time.Sleep(10 * time.Millisecond) // 假设读操作耗时10毫秒rwlock.Unlock()                   // 解写锁// lock.Unlock()                     // 解互斥锁wg.Done()
}func read() {// lock.Lock()                  // 加互斥锁rwlock.RLock()               // 加读锁time.Sleep(time.Millisecond) // 假设读操作耗时1毫秒rwlock.RUnlock()             // 解读锁// lock.Unlock()                // 解互斥锁wg.Done()
}func main() {start := time.Now()for i := 0; i < 10; i++ {wg.Add(1)go write()}for i := 0; i < 1000; i++ {wg.Add(1)go read()}wg.Wait()end := time.Now()fmt.Println(end.Sub(start))
}

3、sync.Once

延迟一个开销很大的初始化操作到真正用到它的时候再执行是一个很好的实践。因为预先初始化一个变量（比如在init函数中完成初始化）会增加程序的启动耗时，而且有可能实际执行过程中这个变量没有用上，那么这个初始化操作就不是必须要做的。

（1）并发控制—执行一次（Java双重检验锁）

在编程的很多场景下我们需要确保某些操作在高并发的场景下只执行一次，例如只加载一次配置文件、只关闭一次通道等。这时我们就需要用到 sync.Once。

sync.Once只有一个Do方法，其签名如下：

// 备注：如果要执行的函数f需要传递参数就需要搭配闭包来使用。
func (o *Once) Do(f func()) {}

接下来我们来看一个并发控制的例子：

var icons map[string]image.Image //value是一个方法func loadIcons() {icons = map[string]image.Image{"left":  loadIcon("left.png"),"up":    loadIcon("up.png"),"right": loadIcon("right.png"),"down":  loadIcon("down.png"),}
}// Icon 被多个goroutine调用时不是并发安全的
func Icon(name string) image.Image {if icons == nil { // 未加载配置文件loadIcons() // 加载}return icons[name]
}

------------------------------------------------------------------------------------------
红线中间不懂，后续研究，先忽略
多个goroutine并发调用Icon函数时不是并发安全的，现代的编译器和CPU可能会在保证每个goroutine都满足串行一致的基础上自由地重排访问内存的顺序。loadIcons函数可能会被重排为以下结果：

func loadIcons() {icons = make(map[string]image.Image)icons["left"] = loadIcon("left.png")icons["up"] = loadIcon("up.png")icons["right"] = loadIcon("right.png")icons["down"] = loadIcon("down.png")
}

------------------------------------------------------------------------------------------
上述，并没有对并发进行任何控制，loadIcons()会进行多次执行，这样性能很差。

下面是 Java并发安全只加载配置文件一次 的示例代码：

import java.io.IOException;
import java.util.Properties;public class PropTool {//volatile关键字是java中禁止指令重排序的关键字，保证有序性和可见性private static volatile Properties prop=null;//出现线程安全问题public static  Properties getProp() throws IOException {if(prop ==null){synchronized ("lock"){if(prop ==null){prop=new Properties();prop.load(LogCollectorTask.class.getClassLoader().getResourceAsStream("collector.properties"));}}}return prop;}}

这是经典的 double check locking（双重检验锁） 场景！
外层判空，因为synchronized加锁很重，为了避免多余的加锁操作
内层判空，是为了防止第一个线程释放锁之后，第二个线程获取到锁，又去加载配置文件。

那么， Go并发安全只加载配置文件一次 是怎么解决的呢？

var icons map[string]image.Imagevar loadIconsOnce sync.Oncefunc loadIcons() {icons = map[string]image.Image{"left":  loadIcon("left.png"),"up":    loadIcon("up.png"),"right": loadIcon("right.png"),"down":  loadIcon("down.png"),}
}// Icon 是并发安全的
func Icon(name string) image.Image {if icons == nil { // 外层判断，避免过多的加锁loadIconsOnce.Do(loadIcons) // sync.once做了加锁与类似的内部判空，为什么叫类似？因为它其实设置了个标志位（0，1）来控制下一个线程不会再加载配置文件}return icons[name]
}

（2）并发安全的单例模式

下面是借助sync.Once实现的并发安全的单例模式：

package singletonimport ("sync"
)type singleton struct {}var instance *singleton
var once sync.Oncefunc GetInstance() *singleton {once.Do(func() {instance = &singleton{}})return instance
}

sync.Once其实内部包含一个互斥锁和一个布尔值，互斥锁保证布尔值和数据的安全，而布尔值用来记录初始化是否完成。这样设计就能保证初始化操作的时候是并发安全的并且初始化操作也不会被执行多次。

4、sync.Map

Go语言中内置的map不是并发安全的。请看下面的示例：

var m = make(map[string]int)func get(key string) int {return m[key]
}func set(key string, value int) {m[key] = value
}func main() {wg := sync.WaitGroup{}for i := 0; i < 20; i++ { // 20是临界值，多了就会报错wg.Add(1)go func(n int) {key := strconv.Itoa(n)set(key, n)fmt.Printf("k=:%v,v:=%v\n", key, get(key))wg.Done()}(i)}wg.Wait()
}

上面的代码开启少量几个goroutine的时候可能没什么问题，当并发多了之后执行上面的代码就会报fatal error: concurrent map writes错误。

像这种场景下就需要为map加锁来保证并发的安全性了，Go语言的sync包中提供了一个开箱即用的并发安全版map–sync.Map。开箱即用表示不用像内置的map一样使用make函数初始化就能直接使用。同时sync.Map内置了诸如Store、Load、LoadOrStore、Delete、Range等操作方法。

var m = sync.Map{}func main() {wg := sync.WaitGroup{}for i := 0; i < 20; i++ {wg.Add(1)go func(n int) {key := strconv.Itoa(n)m.Store(key, n)value, _ := m.Load(key)fmt.Printf("k=:%v,v:=%v\n", key, value)wg.Done()}(i)}wg.Wait()
}

九、生产者消费者（kubernetes队列）

思想：一般来说我们需要生产者和消费者进行交互，生产者再生产之后需要唤醒消费者去消费，且应该判断生产者队列满的情况，消费者队列空的情况，然后进入wait等待。

Kubernetes 中的队列，就是标准的生产者消费者模式

Golang的sync包中的Cond实现了一种条件变量，主要用来解决多个读协程等待共享资源变成ready的场景。在使用Cond的时候，需要特别注意下：每个Cond都会关联一个Lock（*sync.Mutex or *sync.RWMutex），当修改条件或者调用Wait方法时，必须加锁，保护condition。

cond: sync.NewCond(&sync.Mutex{})

Cond相关API介绍, Cond主要有三个函数构成，Broadcast(), Signal(), Wait()。

需要特别注意的是wait操作：

主要由四步构成：1、Unlock()-> // 调用wait，进入等待，因此释放锁，如果有数据，别人被随机唤醒也能执行到相应操作，不能占着茅坑不拉s2、阻塞等待通知(即等待Signal()或Broadcast()的通知)->3、收到通知->4、Lock()源码如下：
func (c *Cond) Wait() {c.checker.check()t := runtime_notifyListAdd(&c.notify)c.L.Unlock()runtime_notifyListWait(&c.notify, t)c.L.Lock()
}

// kubernetes中的队列的实现，标准生产者消费者
package mainimport ("fmt""sync""time"
)type Queue struct {queue []stringcond  *sync.Cond
}func main() {q := Queue{queue: []string{},cond:  sync.NewCond(&sync.Mutex{}),}go func() {for {q.Enqueue("a")time.Sleep(time.Second * 2)}}()for {q.Dequeue()time.Sleep(time.Second)}
}func (q *Queue) Enqueue(item string) {q.cond.L.Lock()defer q.cond.L.Unlock()// 注意因为这里是用切片实现的queue因此不用判断是否满了，阻塞q.queue = append(q.queue, item)fmt.Printf("putting %s to queue, notify all\n", item)// 唤醒所有协程争抢锁q.cond.Broadcast()// 随机唤醒其中一个协程// q.cond.Signal()
}func (q *Queue) Dequeue() string {q.cond.L.Lock()defer q.cond.L.Unlock()for len(q.queue) == 0 {fmt.Println("no data available, wait")q.cond.Wait()}result, q.queue := q.queue[0], q.queue[1:]return result
}

十、原子操作

在上面的代码中的我们通过锁操作来实现同步。而锁机制的底层是基于原子操作的，其一般直接通过CPU指令实现。Go语言中原子操作由内置的标准库sync/atomic提供。

1、atomic包

2、示例（比较互斥锁、原子操作的性能）

package mainimport ("fmt""sync""sync/atomic""time"
)type Counter interface {Inc()Load() int64
}// 普通版
type CommonCounter struct {counter int64
}func (c CommonCounter) Inc() {c.counter++
}func (c CommonCounter) Load() int64 {return c.counter
}// 互斥锁版
type MutexCounter struct {counter int64lock    sync.Mutex
}func (m *MutexCounter) Inc() {m.lock.Lock()defer m.lock.Unlock()m.counter++
}func (m *MutexCounter) Load() int64 {m.lock.Lock()defer m.lock.Unlock()return m.counter
}// 原子操作版
type AtomicCounter struct {counter int64
}func (a *AtomicCounter) Inc() {atomic.AddInt64(&a.counter, 1)
}func (a *AtomicCounter) Load() int64 {return atomic.LoadInt64(&a.counter)
}func test(c Counter) {var wg sync.WaitGroupstart := time.Now()for i := 0; i < 1000; i++ {wg.Add(1)go func() {c.Inc()wg.Done()}()}wg.Wait()end := time.Now()fmt.Println(c.Load(), end.Sub(start))
}func main() {c1 := CommonCounter{} // 非并发安全test(c1)c2 := MutexCounter{} // 使用互斥锁实现并发安全test(&c2)c3 := AtomicCounter{} // 并发安全且比互斥锁效率更高test(&c3)
}

atomic包提供了底层的原子级内存操作，对于同步算法的实现很有用。这些函数必须谨慎地保证正确使用。除了某些特殊的底层应用，使用通道或者sync包的函数/类型实现同步更好