Go并发编程
概述
简而言之，所谓并发编程是指在一台处理器上“同时”处理多个任务。
随着硬件的发展，并发程序变得越来越重要。web服务器会一次处理成千上万的请求。平板电脑和手机app在渲染用户画面同时还会在后台执行各种计算任务和网络请求。即使是传统的批处理问题--读取数据，计算，写输出--现在也会用并发来隐藏掉i/o的操作延迟以充分利用现代计算机设备的多个核心。计算机的性能每年都在以非线性的速度增长。
宏观的并发是指在一段时间内，有多个程序在同时运行。
并发在微观上，是指在同一时刻只能有一条指令执行，但多个程序指令被快速的轮换执行，使得在宏观上具有多个进程同时执行的效果，但在微观上并不是同时执行的，只是把时间分成若干段，使多个程序快速交替的执行。
并行和并发
并行(parallel)：指在同一时刻，有多条指令在多个处理器上同时执行。
并发(concurrency)：指在同一时刻只能有一条指令执行，但多个进程指令被快速的轮换执行，使得在宏观上具有多个进程同时执行的效果，但在微观上并不是同时执行的，只是把时间分成若干段，通过cpu时间片轮转使多个进程快速交替的执行。
大师曾以咖啡机的例子来解释并行和并发的区别。
- 并行是两个队列同时使用两台咖啡机 （真正的多任务）
- 并发是两个队列交替使用一台咖啡机 （ 假 的多任务）
下面的两张图是国人做的一些不错的比较并行与并发区别的图。
常见并发编程技术
进程并发
程序和进程
程序，是指编译好的二进制文件，在磁盘上，不占用系统资源(cpu、内存、打开的文件、设备、锁....)
进程，是一个抽象的概念，与操作系统原理联系紧密。进程是活跃的程序，占用系统资源。在内存中执行。(程序运行起来，产生一个进程)
我们可以把程序比作剧本(纸)，把进程比作一场戏(舞台、演员、灯光、道具...)
同一个剧本可以在多个舞台同时上演。同样，同一个程序也可以加载为不同的进程(彼此之间互不影响)
如：同时开两个终端。各自都有一个bash但彼此id不同。
进程状态
进程基本的状态有5种。分别为初始态，就绪态，，挂起态与终止态。其中初始态为进程准备阶段，常与就绪态结合来看。
进程并发
在使用进程 实现并发时会出现什么问题呢？
- 系统开销比较大，占用资源比较多，开启进程数量比较少。
- 在unix/linux系统下，还会产生“孤儿进程”和“僵尸进程”。
我们知道在操作系统中，可以产生很多的进程。在unix/linux系统中，正常情况下，子进程是通过父进程fork创建的，子进程再创建新的进程。
并且父进程永远无法预测子进程 到底什么时候结束。 当一个 进程完成它的工作终止之后，它的父进程需要调用系统调用取得子进程的终止状态。
孤儿进程
父进程先于子进程结束，则子进程成为孤儿进程，子进程的父进程成为init进程，称为init进程领养孤儿进程。
僵尸进程
进程终止，父进程尚未回收，子进程残留资源（pcb）存放于内核中，变成僵尸（zombie）进程。
windows下的进程和linux下的进程是不一样的，它比较懒惰，从来不执行任何东西，只是为线程提供执行环境。然后由线程负责执行包含在进程的地址空间中的代码。当创建一个进程的时候，操作系统会自动创建这个进程的第一个线程，称为主线程。
线程并发
什么是线程
lwp：light weight process 轻量级的进程，本质仍是进程 (linux下)
进程：独立地址空间，拥有pcb
线程：有独立的pcb，但没有独立的地址空间(共享)
区别：在于是否共享地址空间。独居(进程)；合租(线程)。
- 线程：最小的执行单位
- 进程：最小分配资源单位，可看成是只有一个线程的进程。
windows系统下，可以直接忽略进程的概念，只谈线程。因为线程是最小的执行单位，是被系统独立调度和分派的基本单位。而进程只是给线程提供执行环境。
线程同步
同步即协同步调，按预定的先后次序运行。
线程同步，指一个线程发出某一功能调用时，在没有得到结果之前，该调用不返回。同时其它线程为保证数据一致性，不能调用该功能。
举例1： 银行存款 5000。柜台，折：取3000；提款机，卡：取 3000。剩余：2000
举例2： 内存中100字节，线程t1欲填入全1， 线程t2欲填入全0。但如果t1执行了50个字节失去cpu，t2执行，会将t1写过的内容覆盖。当t1再次获得cpu继续 从失去cpu的位置向后写入1，当执行结束，内存中的100字节，既不是全1，也不是全0。
产生的现象叫做“与时间有关的错误”(time related)。为了避免这种数据混乱，线程需要同步。
“同步”的目的，是为了避免数据混乱，解决与时间有关的错误。实际上，不仅线程间需要同步，进程间、信号间等等都需要同步机制。
因此，所有“多个控制流，共同操作一个共享资源”的情况，都需要同步。
锁的应用
互斥量mutex
linux中提供一把互斥锁mutex（也称之为互斥量）。
每个线程在对资源操作前都尝试先加锁，成功加锁才能操作，操作结束解锁。
资源还是共享的，线程间也还是竞争的，
但通过“锁”就将资源的访问变成互斥操作，而后与时间有关的错误也不会再产生了。
但，应注意：同一时刻，只能有一个线程持有该锁。
当a线程对某个全局变量加锁访问，b在访问前尝试加锁，拿不到锁，b阻塞。c线程不去加锁，而直接访问该全局变量，依然能够访问，但会出现数据混乱。
所以，互斥锁实质上是操作系统提供的一把“建议锁”（又称“协同锁”），建议程序中有多线程访问共享资源的时候使用该机制。但，并没有强制限定。
因此，即使有了mutex，如果有线程不按规则来访问数据，依然会造成数据混乱。
读写锁
与互斥量类似，但读写锁允许更高的并行性。其特性为：写独占，读共享。
读写锁状态：
特别强调：读写锁只有一把，但其具备两种状态：
- 读模式下加锁状态 (读锁)
- 写模式下加锁状态 (写锁)
读写锁特性：
- 读写锁是“写模式加锁”时， 解锁前，所有对该锁加锁的线程都会被阻塞。
- 读写锁是“读模式加锁”时， 如果线程以读模式对其加锁会成功；如果线程以写模式加锁会阻塞。
- 读写锁是“读模式加锁”时， 既有试图以写模式加锁的线程，也有试图以读模式加锁的线程。那么读写锁会阻塞随后的读模式锁请求。优先满足写模式锁。读锁、写锁并行阻塞，写锁优先级高
读写锁也叫共享-独占锁。当读写锁以读模式锁住时，它是以共享模式锁住的；当它以写模式锁住时，它是以独占模式锁住的。写独占、读共享。
读写锁非常适合于对数据结构读的次数远大于写的情况。
协程并发
协程：coroutine。也叫轻量级线程。
与传统的系统级线程和进程相比，协程最大的优势在于“轻量级”。可以轻松创建上万个而不会导致系统资源衰竭。而线程和进程通常很难超过1万个。这也是协程别称“轻量级线程”的原因。
一个线程中可以有任意多个协程，但某一时刻只能有一个协程在运行，多个协程分享该线程分配到的计算机资源。
多数语言在语法层面并不直接支持协程，而是通过库的方式支持，但用库的方式支持的功能也并不完整，比如仅仅提供协程的创建、销毁与切换等能力。如果在这样的轻量级线程中调用一个同步 io 操作，比如网络通信、本地文件读写，都会阻塞其他的并发执行轻量级线程，从而无法真正达到轻量级线程本身期望达到的目标。
在协程中，调用一个任务就像调用一个函数一样，消耗的系统资源最少！但能达到进程、线程并发相同的效果。
在一次并发任务中，进程、线程、协程均可以实现。从系统资源消耗的角度出发来看，进程相当多，线程次之，协程最少。
go并发
go 在语言级别支持协程，叫goroutine。go 语言标准库提供的所有系统调用操作（包括所有同步io操作），都会出让cpu给其他goroutine。这让轻量级线程的切换管理不依赖于系统的线程和进程，也不需要依赖于cpu的核心数量。
有人把go比作21世纪的c语言。第一是因为go语言设计简单，第二，21世纪最重要的就是并行程序设计，而go从语言层面就支持并行。同时，并发程序的内存管理有时候是非常复杂的，而go语言提供了自动垃圾回收机制。
go语言为并发编程而内置的上层api基于顺序通信进程模型csp(communicating sequential processes)。这就意味着显式锁都是可以避免的，因为go通过相对安全的通道发送和接受数据以实现同步，这大大地简化了并发程序的编写。
go语言中的并发程序主要使用两种手段来实现。goroutine和channel。
goroutine
什么是goroutine
goroutine是go并行设计的核心。goroutine说到底其实就是协程，它比线程更小，十几个goroutine可能体现在底层就是五六个线程，go语言内部帮你实现了这些goroutine之间的内存共享。执行goroutine只需极少的栈内存(大概是4~5kb)，当然会根据相应的数据伸缩。也正因为如此，可同时运行成千上万个并发任务。goroutine比thread更易用、更高效、更轻便。
一般情况下，一个普通计算机跑几十个线程就有点负载过大了，但是同样的机器却可以轻松地让成百上千个goroutine进行资源竞争。
goroutine的创建
只需在函数调⽤语句前添加 go 关键字，就可创建并发执⾏单元。开发⼈员无需了解任何执⾏细节，调度器会自动将其安排到合适的系统线程上执行。
在并发编程中，我们通常想将一个过程切分成几块，然后让每个goroutine各自负责一块工作，当一个程序启动时，主函数在一个单独的goroutine中运行，我们叫它main goroutine。新的goroutine会用go语句来创建。而go语言的并发设计，让我们很轻松就可以达成这一目的。
示例代码:
package main
import (
"fmt"
"time"
)
func newtask() {
i := 0
for {
i++
fmt.printf("new goroutine: i = %d\n", i)
time.sleep(time.second) //延时1秒
}
}
func main() {
//创建一个goroutine，启动另外一个任务
go newtask()
//循环打印
for i := 0; i < 5; i++ {
fmt.printf("main goroutine: i = %d\n", i)
time.sleep(time.second) //延时1秒
i++
}
}
程序运行结果:
goroutine特性
主goroutine退出后，其它的工作的子goroutine也会自动退出：
可以看出，由于主goroutine(main函数)执行太快了，所以导致newtask还没执行，程序就退出了
package main
import (
"fmt"
"time"
)
func newtask() {
i := 0
for {
i++
fmt.printf("new goroutine: i = %d\n", i)
time.sleep(time.second) //延时1秒
}
}
func main() {
//创建一个goroutine，启动另外一个任务
go newtask()
fmt.println("hello world")
}
程序运行结果:
通过运行结果(运行了三次)可以看出来，其中有一次newtask得到了执行，但是也只输出了一次程序就退出了。另外两次newtask完全完全没有执行就退出程序了。
runtime包
gosched
runtime.gosched() 用于让出cpu时间片，让出当前goroutine的执行权限，调度器安排其他等待的任务运行，并在下次再获得cpu时间轮片的时候，从该出让cpu的位置恢复执行。
有点像跑接力赛，a跑了一会碰到代码runtime.gosched() 就把接力棒交给b了，a歇着了，b继续跑。
示例代码:
package main
import (
"fmt"
"runtime"
)
func main() {
//创建一个goroutine
go func(s string) {
for i := 0; i < 2; i++ {
fmt.println(s)
}
}("world")
for i := 0; i < 2; i++ {
runtime.gosched() //import "runtime"包
fmt.println("hello")
}
/*
屏蔽runtime.gosched()运行结果如下:
hello
hello
没有runtime.gosched()运行结果如下:
world
world
hello
hello
*/
}
以上程序的执行过程如下：
主协程进入main()函数，进行代码的执行。当执行到go func()匿名函数时，创建一个新的协程，开始执行匿名函数中的代码，主协程继续向下执行，执行到runtime.gosched( )时会暂停向下执行，直到其它协程执行完后，再回到该位置，主协程继续向下执行。
goexit
调用 runtime.goexit() 将立即终止当前 goroutine 执⾏，调度器确保所有已注册 defer延迟调用被执行。
示例代码:
package main
import (
"fmt"
"runtime"
)
func main() {
go func() {
defer fmt.println("a.defer")
func() {
defer fmt.println("b.defer")
runtime.goexit()//终止当前 goroutine, import "runtime"
fmt.println("b") //不会执行
}()
fmt.println("a") //不会执行
}()
//死循环，目的不让主goroutine结束
for {}
}
程序运行结果:
gomaxprocs
调用 runtime.gomaxprocs() 用来设置可以并行计算的cpu核数的最大值，并返回之前的值。(默认是跑满整个cpu)
示例代码:
package main
import (
"fmt"
"runtime"
)
func main() {
//n := runtime.gomaxprocs(1) //第一次 测试
//打印结果: 111111111111111111111111110000000000000000000000000....
n := runtime.gomaxprocs(2) //第二次 测试
//打印结果: 1111111111111111111111110000000000000011111110000100000000111100001111
fmt.println(n)
for {
go fmt.print(0)
fmt.print(1)
}
}
在第一次执行runtime.gomaxprocs(1) 时，最多同时只能有一个goroutine被执行。所以会打印很多1。过了一段时间后，go调度器会将其置为休眠，并唤醒另一个goroutine，这时候就开始打印很多0了，在打印的时候，goroutine是被调度到操作系统线程上的。
在第二次执行runtime.gomaxprocs(2) 时， 我们使用了两个cpu，所以两个goroutine可以一起被执行，以同样的频率交替打印0和1。
channel
channel是go语言中的一个核心类型，可以把它看成管道。并发核心单元通过它就可以发送或者接收数据进行通讯，这在一定程度上又进一步降低了编程的难度。
channel是一个数据类型，主要用来解决协程的同步问题以及协程之间数据共享（数据传递）的问题。
goroutine运行在相同的地址空间，因此访问共享内存必须做好同步。goroutine 奉行通过通信来共享内存，而不是共享内存来通信。
引⽤类型 channel可用于多个 goroutine 通讯。其内部实现了同步，确保并发安全。
定义channel变量
和map类似，channel也一个对应make创建的底层数据结构的引用。
当我们复制一个channel或用于函数参数传递时，我们只是拷贝了一个channel引用，因此调用者和被调用者将引用同一个channel对象。和其它的引用类型一样，channel的零值也是nil。
定义一个channel时，也需要定义发送到channel的值的类型。channel可以使用内置的make()函数来创建：
chan是创建channel所需使用的关键字。type 代表指定channel收发数据的类型。
make(chan type) //等价于make(chan type, 0) make(chan type, capacity)
当我们复制一个channel或用于函数参数传递时，我们只是拷贝了一个channel引用，因此调用者和被调用者将引用同一个channel对象。和其它的引用类型一样，channel的零值也是nil。
当 参数capacity= 0 时，channel 是无缓冲阻塞读写的；当capacity > 0 时，channel 有缓冲、是非阻塞的，直到写满 capacity个元素才阻塞写入。
channel非常像生活中的管道，一边可以存放东西，另一边可以取出东西。channel通过操作符 <- 来接收和发送数据，发送和接收数据语法：
channel <- value //发送value到channel <- channel //取出channel里的一个值并丢弃 x := <-channel //从channel中接收数据，并赋值给x x, ok := <-channel //功能同上，同时检查通道是否已关闭或者是否为空
默认情况下，无缓冲的channel接收和发送数据都是阻塞的，除非另一端已经准备好，这样就使得goroutine同步变的更加的简单，而不需要显式的lock。
示例代码:
package main
import "fmt"
func main() {
c := make(chan int)
go func() {
defer fmt.println("子协程结束")
fmt.println("子协程正在运行.....")
c <- 666 //666发送到c
}()
num := <-c //从c中接收数据，并赋值给num
fmt.println("num = ", num)
fmt.println("main协程结束")
}
程序运行结果:
无缓冲的channel
无缓冲的通道（unbuffered channel）是指在接收前没有能力保存任何值的通道。
这种类型的通道要求发送goroutine和接收goroutine同时准备好，才能完成发送和接收操作。否则，通道会导致先执行发送或接收操作的 goroutine 阻塞等待。
这种对通道进行发送和接收的交互行为本身就是同步的。其中任意一个操作都无法离开另一个操作单独存在。
阻塞：由于某种原因数据没有到达，当前协程（线程）持续处于等待状态，直到条件满足，才接触阻塞。
同步：在两个或多个协程（线程）间，保持数据内容一致性的机制。
下图展示两个 goroutine 如何利用无缓冲的通道来共享一个值：
- 在第 1 步，两个 goroutine 都到达通道，但哪个都没有开始执行发送或者接收。
- 在第 2 步，左侧的 goroutine 将它的手伸进了通道，这模拟了向通道发送数据的行为。这时，这个 goroutine 会在通道中被锁住，直到交换完成。
- 在第 3 步，右侧的 goroutine 将它的手放入通道，这模拟了从通道里接收数据。这个 goroutine 一样也会在通道中被锁住，直到交换完成。
- 在第 4 步和第 5 步，进行交换，并最终，在第 6 步，两个 goroutine 都将它们的手从通道里拿出来，这模拟了被锁住的 goroutine 得到释放。两个 goroutine 现在都可以去做别的事情了。
无缓冲的channel创建格式：
make(chan type) //等价于make(chan type, 0)
如果没有指定缓冲区容量，那么该通道就是同步的，因此会阻塞到发送者准备好发送和接收者准备好接收。
示例代码:
package main
import (
"fmt"
"time"
)
func main() {
c := make(chan int, 0) //创建无缓冲的通道c
//内置函数len返回未被读取的缓冲元素数量，cap返回缓冲区大小
fmt.printf("len(c) = %d, cap(c) = %d\n", len(c), cap(c))
go func() {
defer fmt.println("子协程结束")
for i := 0; i < 3; i++ {
c <- i
fmt.printf("子协程正在运行[%d]: len(c) = %d, cap(c) = %d\n", i, len(c), cap(c))
}
}()
time.sleep(time.second * 2) //延时2s
for i := 0; i < 3; i++ {
num := <-c //从c中接收数据，并复制给num
fmt.printf("num = %d\n", num)
}
fmt.println("main协程结束")
}
有缓冲的channel
有缓冲的通道（buffered channel）是一种在被接收前能存储一个或者多个数据值的通道。
这种类型的通道并不强制要求 goroutine 之间必须同时完成发送和接收。通道会阻塞发送和接收动作的条件也不同。
只有通道中没有可以接收的值时，接收动作才会阻塞。
只有通道没有可用缓冲区容纳被发送的值时，发送动作才会阻塞。
这导致有缓冲的通道和无缓冲的通道之间的一个很大的不同：无缓冲的通道保证进行发送和接收的 goroutine 会在同一时间进行数据交换；有缓冲的通道没有这种保证。
示例如下:
- 在第 1 步，右侧的 goroutine 正在从通道接收一个值。
- 在第 2 步，右侧的这个 goroutine独立完成了接收值的动作，而左侧的 goroutine 正在发送一个新值到通道里。
- 在第 3 步，左侧的goroutine 还在向通道发送新值，而右侧的 goroutine 正在从通道接收另外一个值。这个步骤里的两个操作既不是同步的，也不会互相阻塞。
- 最后，在第 4 步，所有的发送和接收都完成，而通道里还有几个值，也有一些空间可以存更多的值。
有缓冲的channel创建格式：
make(chan type, capacity)
如果给定了一个缓冲区容量，通道就是异步的。只要缓冲区有未使用空间用于发送数据，或还包含可以接收的数据，那么其通信就会无阻塞地进行。
示例代码:
package main
import (
"fmt"
"time"
)
func main() {
c := make(chan int, 3) //创建无缓冲的通道c
//内置函数len返回未被读取的缓冲元素数量，cap返回缓冲区大小
fmt.printf("len(c) = %d, cap(c) = %d\n", len(c), cap(c))
go func() {
defer fmt.println("子协程结束")
for i := 0; i < 3; i++ {
c <- i
fmt.printf("子协程正在运行[%d]: len(c) = %d, cap(c) = %d\n", i, len(c), cap(c))
}
}()
time.sleep(time.second * 2) //延时2s
for i := 0; i < 3; i++ {
num := <-c //从c中接收数据，并复制给num
fmt.printf("num = %d\n", num)
}
fmt.println("main协程结束")
}
关闭channel
如果发送者知道，没有更多的值需要发送到channel的话，那么让接收者也能及时知道没有多余的值可接收将是有用的，因为接收者可以停止不必要的接收等待。这可以通过内置的close函数来关闭channel实现。
示例代码:
注意:
- channel不像文件一样需要经常去关闭，只有当你确实没有任何发送数据了，或者你想显式的结束range循环之类的，才去关闭channel；
- 关闭channel后，无法向channel 再发送数据(引发 panic 错误后导致接收立即返回channel类型的零值)；
- 关闭channel后，可以继续从channel接收数据；
- 对于nil channel，无论收发都会被阻塞。
可以使用 range 来迭代不断操作channel：
package main
import "fmt"
func main() {
c := make(chan int)
go func() {
for i := 0; i < 5; i++ {
c <- i
}
//把close(c)注释掉，程序会死锁
//close(c)
}()
for {
//ok为true说明channel没有关闭，为false说明channel已经关闭
if data, ok := <-c; ok {
fmt.println(data)
} else {
break
}
}
fmt.println("main协程结束")
}
单向channel
有的时候我们会将通道作为参数在多个任务函数间传递，很多时候我们在不同的任务函数中使用通道都会对其进行限制，比如限制通道在函数中只能发送或只能接收。
go语言中提供了单向通道来处理这种情况。例如，我们把上面的例子改造如下：
package main
import (
"fmt"
)
func counter(in chan<- int) {
for i := 0; i < 100; i++ {
in <- i
}
close(in)
}
func squarer(in chan<- int, out <-chan int) {
for i := range out {
in <- i * i
}
close(in)
}
func printer(in <-chan int) {
for i := range in {
fmt.println(i)
}
}
func main() {
ch1 := make(chan int)
ch2 := make(chan int)
go counter(ch1)
go squarer(ch2, ch1)
printer(ch2)
}
其中
-
chan<- int是一个只能发送的通道，可以发送但是不能接收； -
<-chan int是一个只能接收的通道，可以接收但是不能发送。
在函数传参及任何赋值操作中将双向通道转换为单向通道是可以的，但反过来是不可以的。
channel总结
定时器
time.timer
timer是一个定时器，代表未来的一个单一事件，你可以告诉timer你需要等待多长事件，它提供一个channel，在将来的那个时间那个channel提供了一个时间值。
它提供一个channel，在定时时间到达之前，没有数据写入timer.c会一直阻塞。直到定时时间到，向channel写入值，阻塞解除，可以从中读取数据。
示例代码:
package main
import (
"fmt"
"time"
)
func main() {
//创建定时器,2秒后,定时器就会向自己的c字段发送一个time.time类型的元素值
timer1 := time.newtimer(time.second * 2)
t1 := time.now() //当前时间
fmt.printf("t1: %v\n", t1)
t2 := <-timer1.c
fmt.printf("t2: %v\n", t2)
//如果只是想单纯的等待的话,可以使用time.sleep来实现
timer2 := time.newtimer(time.second * 2)
<-timer2.c
fmt.println("2s后")
time.sleep(time.second * 2)
fmt.println("再一次2s后")
<-time.after(time.second * 2)
fmt.println("再再一次2s后")
timer3 := time.newtimer(time.second)
go func() {
<-timer3.c
fmt.println("timer 3 expired")
}()
stop := timer3.stop() //停止定时器
if stop {
fmt.println("timer 3 stopped")
}
fmt.println("before")
timer4 := time.newtimer(time.second * 5) //原来设置5秒
timer4.reset(time.second * 1) //重新设置时间
<-timer4.c
fmt.println("after")
}
定时器的常用操作:
- 实现延迟功能
(1) <-time.after(time.second * 2) //定时2s,阻塞2s, 2s后产生一个事件，往channel写内容
fmt.println("时间到") (2) time.sleep(time.second * 2)
fmt.println("时间到")(3) 延时2s后打印一句话
timer := time.newtimer(time.second * 2)
<-timer.c
fmt.println("时间到") -
定时器停止
timer := time.newtimer(time.second * 3) go func() { <-timer.c fmt.println("子协程可以打印了,因为定时器的时间到") }() timer.stop() //停止定时器 for { } -
定时器重置
timer := time.newtimer(3 * time.second) ok := timer.reset(1 * time.second) //重新设置为1s fmt.println("ok = ", ok) <-timer.c fmt.println("时间到")
time.ticker
ticker是一个定时触发的计时器，它会以一个间隔(interval)往channel发送一个事件(当前时间)，而channel的接受者可以以固定的时间间隔从channel中读取事件。
示例代码:
package main
import (
"fmt"
"time"
)
func main() {
//创建定时器，每隔1s后，定时器就会给channel发送一个事件(当前时间)
ticker := time.newticker(1 * time.second)
go func() {
i := 0
for { //循环
<-ticker.c
i++
fmt.println("i =", i)
if i == 5 {
ticker.stop() //停止定时器
}
}
}()
//死循环，特地不让main goroutine结束
for {
}
}
select
select作用
go里面提供了一个关键字select，通过select可以监听channel上的数据流动。
select的用法与switch语言非常类似，由select开始一个新的选择块，每个选择条件由case语句来描述。
与switch语句相比， select有比较多的限制，其中最大的一条限制就是每个case语句里必须是一个io操作，大致的结构如下：
select {
case <-chan1:
// 如果chan1成功读到数据，则进行该case处理语句
case chan2 <- 1:
// 如果成功向chan2写入数据，则进行该case处理语句
default:
// 如果上面都没有成功，则进入default处理流程
}
在一个select语句中，go语言会按顺序从头至尾评估每一个发送和接收的语句。
如果其中的任意一语句可以继续执行(即没有被阻塞)，那么就从那些可以执行的语句中任意选择一条来使用。
如果没有任意一条语句可以执行(即所有的通道都被阻塞)，那么有两种可能的情况：
- 如果给出了default语句，那么就会执行default语句，同时程序的执行会从select语句后的语句中恢复。
- 如果没有default语句，那么select语句将被阻塞，直到至少有一个通信可以进行下去。
示例代码:
package main
import "fmt"
func fibonacci(c, quit chan int) {
x, y := 1, 1
for {
select {
case c <-x:
x, y = y, x+y
case <-quit:
fmt.println("quit")
return
}
}
}
func main() {
c := make(chan int)
quit := make(chan int)
go func() {
for i := 0; i < 6; i++ {
fmt.println(<-c)
}
quit <- 0
}()
fibonacci(c, quit)
}
运行结果如下:
示例2:
package main
import "fmt"
func main() {
var 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函数。
超时
有时候会出现goroutine阻塞的情况，那么我们如何避免整个程序进入阻塞的情况呢？我们可以利用select来设置超时，通过如下的方式实现：
package main
import (
"fmt"
"time"
)
func main() {
c := make(chan int)
o := make(chan bool)
go func() {
for {
select {
case v := <-c:
fmt.println(v)
case <-time.after(5 * time.second):
fmt.println("timeout")
o <- true
break
}
}
}()
//c <- 666 //注释掉，引发timeout
<-o
}
锁
前面我们为了解决协程同步的问题我们使用了channel，但是go也提供了传统的同步工具。
它们都在go的标准库代码包sync和sync/atomic中。
什么是锁呢？就是某个协程（线程）在访问某个资源时先锁住，防止其它协程的访问，等访问完毕解锁后其他协程再来加锁进行访问。这和我们生活中加锁使用公共资源相似，例如：公共卫生间。
死锁
死锁是指两个或两个以上的进程在执行过程中，由于竞争资源或者由于彼此通信而造成的一种阻塞的现象，若无外力作用，它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁
示例代码:
package main
import "fmt"
func main() {
ch := make(chan int)
ch <- 1 // i'm blocked because there is no channel read yet.
fmt.println("send")
go func() {
<-ch // i will never be called for the main routine is blocked!
fmt.println("received")
}()
fmt.println("over")
}
互斥锁
每个资源都对应于一个可称为 "互斥锁" 的标记，这个标记用来保证在任意时刻，只能有一个协程（线程）访问该资源。其它的协程只能等待。
互斥锁是传统并发编程对共享资源进行访问控制的主要手段，它由标准库sync中的mutex结构体类型表示。sync.mutex类型只有两个公开的指针方法，lock和unlock。lock锁定当前的共享资源，unlock进行解锁。
在使用互斥锁时，一定要注意：对资源操作完成后，一定要解锁，否则会出现流程执行异常，死锁等问题。
有时候在go代码中可能会存在多个goroutine同时操作一个资源（临界区），这种情况会发生竞态问题（数据竞态）。类比现实生活中的例子有十字路口被各个方向的的汽车竞争；还有火车上的卫生间被车厢里的人竞争。
举个例子：
package main
import (
"fmt"
"sync"
)
var (
x int64
wg sync.waitgroup
)
func add() {
defer wg.done()
for i := 0; i < 5000; i++ {
x = x + 1
}
}
func main() {
wg.add(2)
go add()
go add()
wg.wait()
fmt.println(x)
}
上面的代码中我们开启了两个goroutine去累加变量x的值，这两个goroutine在访问和修改x变量的时候就会存在数据竞争，导致最后的结果与期待的不符。
互斥锁是一种常用的控制共享资源访问的方法，它能够保证同时只有一个goroutine可以访问共享资源。go语言中使用sync包的mutex类型来实现互斥锁。 使用互斥锁来修复上面代码的问题：
package main
import (
"fmt"
"sync"
)
var (
x int64
wg sync.waitgroup
lock sync.mutex
)
func add() {
defer wg.done()
for i := 0; i < 5000; i++ {
lock.lock() //加锁
x = x + 1
lock.unlock() //解锁
}
}
func main() {
wg.add(2)
go add()
go add()
wg.wait()
fmt.println(x)
}
使用互斥锁能够保证同一时间有且只有一个goroutine进入临界区，其他的goroutine则在等待锁；当互斥锁释放后，等待的goroutine才可以获取锁进入临界区，多个goroutine同时等待一个锁时，唤醒的策略是随机的。
读写锁
互斥锁的本质是当一个goroutine访问的时候，其他goroutine都不能访问。这样在资源同步，避免竞争的同时也降低了程序的并发性能。程序由原来的并行执行变成了串行执行。
其实，当我们对一个不会变化的数据只做“读”操作的话，是不存在资源竞争的问题的。因为数据是不变的，不管怎么读取，多少goroutine同时读取，都是可以的。
所以问题不是出在“读”上，主要是修改，也就是“写”。修改的数据要同步，这样其他goroutine才可以感知到。所以真正的互斥应该是读取和修改、修改和修改之间，读和读是没有互斥操作的必要的。
因此，衍生出另外一种锁，叫做读写锁。
读写锁可以让多个读操作并发，同时读取，但是对于写操作是完全互斥的。也就是说，当一个goroutine进行写操作的时候，其他goroutine既不能进行读操作，也不能进行写操作。
go中的读写锁由结构体类型sync.rwmutex表示。此类型的方法集合中包含两对方法：
一组是对写操作的锁定和解锁，简称“写锁定”和“写解锁”：
func (*rwmutex)lock() func (*rwmutex)unlock()
另一组表示对读操作的锁定和解锁，简称为“读锁定”与“读解锁”：
func (*rwmutex)rlock() func (*rwmutex)rulock()
读写锁基本示例:
package main
import (
"fmt"
"sync"
"time"
)
var (
x int64
wg sync.waitgroup
rwlock sync.rwmutex
)
func write() {
defer wg.done()
rwlock.lock() //加写锁
x = x + 1
time.sleep(time.millisecond * 10) //假设写操作耗时10毫秒
rwlock.unlock() //解写锁
}
func read() {
defer wg.done()
rwlock.rlock() //加读锁
time.sleep(time.millisecond) //假设读操作耗时一毫秒
rwlock.runlock() //解读锁
}
func main() {
var 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()
var end = time.now()
fmt.println(end.sub(start))
}
我们在read里使用读锁，也就是rlock和runlock，写锁的方法名和我们平时使用的一样，是lock和unlock。这样，我们就使用了读写锁，可以并发地读，但是同时只能有一个写，并且写的时候不能进行读操作。
需要注意的是读写锁非常适合读多写少的场景，如果读和写的操作差别不大，读写锁的优势就发挥不出来。
总结：读写锁控制下的多个写操作之间都是互斥的，并且写操作与读操作之间也都是互斥的。但是，多个读操作之间不存在互斥关系。
从互斥锁和读写锁的源码可以看出，它们是同源的。读写锁的内部用互斥锁来实现写锁定操作之间的互斥。可以把读写锁看作是互斥锁的一种扩展。
sync.waitgroup
在代码中生硬的使用time.sleep肯定是不合适的，go语言中可以使用sync.waitgroup来实现并发任务的同步。 sync.waitgroup有以下几个方法：
| 方法名 | 功能 |
|---|---|
| (wg *waitgroup) add(delta int) | 计数器+delta |
| (wg *waitgroup) done() | 计数器-1 |
| (wg *waitgroup) wait() | 阻塞直到计数器变为0 |
sync.waitgroup内部维护着一个计数器，计数器的值可以增加和减少。例如当我们启动了n 个并发任务时，就将计数器值增加n。每个任务完成时通过调用done()方法将计数器减1。通过调用wait()来等待并发任务执行完，当计数器值为0时，表示所有并发任务已经完成。
我们利用sync.waitgroup将上面的代码优化一下：
package main
import (
"fmt"
"sync"
)
func sayhello(wg *sync.waitgroup) {
defer wg.done()
fmt.println("hello")
}
func main() {
var wg sync.waitgroup
for i := 0; i < 5; i++ {
wg.add(1)
go sayhello(&wg)
}
fmt.println("main goroutine done!")
wg.wait()
}
需要注意sync.waitgroup是一个结构体，传递的时候要传递指针。
sync.once
说在前面的话：这是一个进阶知识点。
在编程的很多场景下我们需要确保某些操作在高并发的场景下只执行一次，例如只加载一次配置文件、只关闭一次通道等。
go语言中的sync包中提供了一个针对只执行一次场景的解决方案–sync.once。
sync.once只有一个do方法，其签名如下：
func (o *once) do(f func()) {}
备注：如果要执行的函数f需要传递参数就需要搭配闭包来使用。
加载配置文件示例
延迟一个开销很大的初始化操作到真正用到它的时候再执行是一个很好的实践。因为预先初始化一个变量（比如在init函数中完成初始化）会增加程序的启动耗时，而且有可能实际执行过程中这个变量没有用上，那么这个初始化操作就不是必须要做的。我们来看一个例子：
var icons map[string]image.image
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")
}
在这种情况下就会出现即使判断了icons不是nil也不意味着变量初始化完成了。考虑到这种情况，我们能想到的办法就是添加互斥锁，保证初始化icons的时候不会被其他的goroutine操作，但是这样做又会引发性能问题。
使用sync.once改造的示例代码如下：
var icons map[string]image.image
var loadiconsonce sync.once
func 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 {
loadiconsonce.do(loadicons)
return icons[name]
}
并发安全的单利模式
下面是借助sync.once实现的并发安全的单例模式：
package main
import "sync"
type singleton struct{}
var (
instance *singleton
once sync.once
)
func getinstance() *singleton {
once.do(func() {
instance = &singleton{}
})
return instance
}
sync.once其实内部包含一个互斥锁和一个布尔值，互斥锁保证布尔值和数据的安全，而布尔值用来记录初始化是否完成。这样设计就能保证初始化操作的时候是并发安全的并且初始化操作也不会被执行多次。
sync.map
go语言中内置的map不是并发安全的。请看下面的示例：
package main
import (
"fmt"
"strconv"
"sync"
)
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() {
var wg = sync.waitgroup{}
for i := 0; i < 20; i++ {
wg.add(1)
go func(n int) {
defer wg.done()
key := strconv.itoa(n)
set(key, n)
fmt.printf("k: %v, v: %v\n", key, get(key))
}(i)
}
wg.wait()
}
上面的代码开启少量几个goroutine的时候可能没什么问题，当并发多了之后执行上面的代码就会报fatal error: concurrent map writes错误。
像这种场景下就需要为map加锁来保证并发的安全性了，go语言的sync包中提供了一个开箱即用的并发安全版map–sync.map。开箱即用表示不用像内置的map一样使用make函数初始化就能直接使用。同时sync.map内置了诸如store、load、loadorstore、delete、range等操作方法。
package main
import (
"fmt"
"strconv"
"sync"
)
func main() {
var m = sync.map{}
var wg = sync.waitgroup{}
for i := 0; i < 40; i++ {
wg.add(1)
go func(n int) {
defer wg.done()
key := strconv.itoa(n)
m.store(key, n)
value, _ := m.load(key)
fmt.printf("k: %v, v: %v\n", key, value)
}(i)
}
wg.wait()
}
原子操作
代码中的加锁操作因为涉及内核态的上下文切换会比较耗时、代价比较高。针对基本数据类型我们还可以使用原子操作来保证并发安全，因为原子操作是go语言提供的方法它在用户态就可以完成，因此性能比加锁操作更好。go语言中原子操作由内置的标准库sync/atomic提供。
atomic包
| 方法 | 解释 |
|---|---|
| func loadint32(addr int32) (val int32) func loadint64(addr int64) (val int64) func loaduint32(addr uint32) (val uint32) func loaduint64(addr uint64) (val uint64) func loaduintptr(addr uintptr) (val uintptr) func loadpointer(addr unsafe.pointer) (val unsafe.pointer) |
读取操作 |
| func storeint32(addr int32, val int32) func storeint64(addr int64, val int64) func storeuint32(addr uint32, val uint32) func storeuint64(addr uint64, val uint64) func storeuintptr(addr uintptr, val uintptr) func storepointer(addr unsafe.pointer, val unsafe.pointer) |
写入操作 |
| func addint32(addr int32, delta int32) (new int32) func addint64(addr int64, delta int64) (new int64) func adduint32(addr uint32, delta uint32) (new uint32) func adduint64(addr uint64, delta uint64) (new uint64) func adduintptr(addr *uintptr, delta uintptr) (new uintptr) |
修改操作 |
| func swapint32(addr int32, new int32) (old int32) func swapint64(addr int64, new int64) (old int64) func swapuint32(addr uint32, new uint32) (old uint32) func swapuint64(addr uint64, new uint64) (old uint64) func swapuintptr(addr uintptr, new uintptr) (old uintptr) func swappointer(addr unsafe.pointer, new unsafe.pointer) (old unsafe.pointer) |
交换操作 |
| func compareandswapint32(addr int32, old, new int32) (swapped bool) func compareandswapint64(addr int64, old, new int64) (swapped bool) func compareandswapuint32(addr uint32, old, new uint32) (swapped bool) func compareandswapuint64(addr uint64, old, new uint64) (swapped bool) func compareandswapuintptr(addr uintptr, old, new uintptr) (swapped bool) func compareandswappointer(addr unsafe.pointer, old, new unsafe.pointer) (swapped bool) |
比较并交换操作 |
示例
我们通过一个示例来比较下互斥锁和原子操作的性能。
package main
import (
"fmt"
"sync"
"sync/atomic"
"time"
)
type counter interface {
inc()
load() int64
}
//普通版
type normalcounter struct {
x int64
}
func (normal *normalcounter) inc() {
normal.x++
}
func (normal *normalcounter) load() int64 {
return normal.x
}
//互斥锁版
type mutexcounter struct {
x int64
lock sync.mutex
}
func (m *mutexcounter) inc () {
m.lock.lock()
m.x++
m.lock.unlock()
}
func (m *mutexcounter) load() int64 {
m.lock.lock()
defer m.lock.unlock()
return m.x
}
//原子操作版
type atomiccounter struct {
x int64
}
func (a *atomiccounter) inc() {
atomic.addint64(&a.x, 1)
}
func (a *atomiccounter) load() int64 {
return atomic.loadint64(&a.x)
}
func test(c counter) {
var wg sync.waitgroup
var start = time.now()
for i := 0; i < 1000; i++ {
wg.add(1)
go func() {
defer wg.done()
c.inc()
}()
}
wg.wait()
var end = time.now()
fmt.printf("执行用时: %v, 结果为: %v\n", end.sub(start), c.load())
}
func main() {
var c1 = normalcounter{} // 非并发安全
test(&c1)
var c2 = mutexcounter{} // 使用互斥锁实现并发安全
test(&c2)
var c3 = atomiccounter{} // 并发安全且比互斥锁效率更高
test(&c3)
}
atomic包提供了底层的原子级内存操作，对于同步算法的实现很有用。这些函数必须谨慎地保证正确使用。除了某些特殊的底层应用，使用通道或者sync包的函数/类型实现同步更好。