golang并发模型
- go在语言层面提供了内置的并发支持
- 不要通过共享内存来通信,而应该通过通信来共享内存
并发与并行
- 定义
- 并发: 指同一时刻, 系统通过调度,来回切换交替的运行多个任务,看起来是"同时"进行的.一个处理器同时处理多个任务
- 并行:指同一时刻,2个任务"真正的"同时进行.多个处理器或者多核的处理器同时处理多个不同的任务.
- 并行:多核cpu,物理上的同时执行.
- 并发:同一时刻,只能一条指令执行.通过快速轮换执行,宏观上是多个线程同时执行的.微观上不是同时执行的,只是把时间分成若干段,使得多个线程快速交替执行.(单核cpu,逻辑上的同时执行)
常见的并发编程模型
- 进程&线程(Apache)
最初的web服务器都是基于进程和线程,比如Apache,新到的一个请求就会分配一个进程或者线程,每个进程只服务一个用户,早期的互联网还不够普及,用户也不够多,这时候网站是可以稳定的,问题是
进程很昂贵,一台服务器无法创建很多的进程,后来随着互联网的发展,用户越来越多,
网站也变得越来越复杂,一个页面有可能就有上百个请求,
所以就诞生了C10K的问题,C10K的意思就是服务器同时支持一个10k量级的并发连接,就是要创建1w个进程,
这样的话,操作系统肯定是无法承受的.
所以进程和线程模型就显得很力不从心了.
- 异步非阻塞(Nginx)
为了解决c10k的问题,就发明了异步非阻塞这种技术,一个典型的案例就是linux里的epoll,
一个普通的服务器就能服务大量的用户,资源消耗也很低,像NGINX都是epoll的产物,
但是,异步非阻塞也不是很完美,为了追求性能,强行的将线性程序打乱,开发和维护都变得非常复杂,
调试起来也比较困难.
- 协程(Golang)
为了降低开发的复杂度,让程序员同学更爽的写代码,协程这种并发模型就逐渐流行起来,
这种模型,可以让我们像写线性程序一样,来写异步的程序,
其实协程的底层就是线程,但它比线程更轻量,几十个协程,体现在底层,可能也就是五六个的线程.
大家把协程理解成,更高效,更易用,更轻量的线程.
Glang并发的实现
- 程序并发执行(goroutine)
每开启一个协程,就会有一个goroutine,负责程序的并发执行
f1() // 执行函数f1,等待函数f1返回
go f1() // 执行函数f1
f2() // 不用等待f1()的返回
使用起来很简单, 只要在一个函数前面加 go 关键字就会创建一个goroutine, 去并发的执行,
所以程序并不会阻塞, 最后这2个函数相当于会去并发的执行.
有了goroutine之后就可以并发的去执行了,这就引出了一个问题:在多个goroutine之间是如何进行数据通信的呢?
比如 go f1() 和 f2() 这2个函数之间要通信,要传递数据的话,那他们是怎么进行的?
使用channel在多个goroutine之间进行数据通信和同步的
- 多个goroutine间的数据同步和通信(channel)
- channel的基础语法
- 创建1:make(chan [type]) // 无缓冲
- 创建2:make(chan [type], int) // 有缓冲
- 写入:channel <-
- 获取: <- channel
- channel的基础语法
c := make(chan string) // 声明一个无缓冲的channel
// 创建一个goroutine
go func() {
c <- "this is channel msg " // 发送数据到channel里
}()
msg := <-c // 阻塞直到接收到数据
fmt.Println(msg)
说明:
// channel分为无缓冲信道(即unbuffered channel)和有缓冲信道(buffered channel)。
无缓冲的与有缓冲channel有着重大差别:一个是同步的 一个是非同步的
ch1:=make(chan int) 无缓冲
ch2:=make(chan int,1) 有缓冲
ch1<-1 // 无缓冲的
这里要有别的协程一直<-ch1 接受这个参数,
那么ch1<-1之后的代码才能执行,要不然就一直阻塞着,
ch2<-1 // 有缓冲的
这里则不会阻塞,因为缓冲大小是1(放了一个缓冲就剩0了),只有当放第二个的时候,第一个还没被拿走,这时候才会阻塞.
比喻:
1. 无缓冲的就是一个送信人去你家门口送信,你不在家他不走,一定要送到你手里他才走.
无缓冲保证信能到你手上.
2. 有缓冲的就是一个送信人去你家门口送信,扔到你家信箱转身就走,除非你的信箱满了,他必须等信箱空下来.
有缓冲保证信能到你家信箱
- 多个channel选择数据读取或者写入(select)
使用select关键字,完成“多路选择”与“超时控制”。
select {
case v := <-ch1:
fmt.Println("channel 1 msg =>", v)
case v := <-ch2:
fmt.Println("channel 2 msg =>", v)
case <-time.After(time.Millisecond * 100): // 超时等待
fmt.Println("time out")
//default:
// fmt.Println("nothing")
}
使用场景:可以监听写信号,进程的热启动,配置的热加载等
协程的使用
func TestGroutine(t *testing.T) {
for i := 0; i < 10; i++ {
go func(i int) {
fmt.Println(i) // 正确案例,值传递。各个协程无竞争关系。
}(i)
// go func() {
// fmt.Println(i) // 错误案例,共享变量。各个协程有竞争关系, 不安全
// }()
}
time.Sleep(time.Millisecond * 50)
}
- 协程并发,导致协程不安全
// 协程不安全demo
func TestThreadUnsafe(t *testing.T) {
counter := 0
for i := 0; i < 5000; i++ {
go func() {
counter++
}()
}
time.Sleep(1 * time.Second)
t.Logf("counter = %d", counter)
}
// 输出结果如下:
=== RUN TestThreadUnsafe
channel_test.go:346: counter = 4742 // 计算错误,因为并发导致了漏值
--- PASS: TestThreadUnsafe (1.00s)
- 如何保证协程安全?
- 方式1: 普通加锁,并延迟等待协程执行完毕(不推荐)
// 协程等待demo(停1秒,不推荐)
func TestThreadSafe(t *testing.T) {
var mut sync.Mutex // 互斥锁
counter := 0
for i := 0; i < 5000; i++ {
go func() { // 开启协程
defer func() {
mut.Unlock() //函数调用完成后:解锁,保证协程安全
}()
mut.Lock() // 函数将要调用前:加锁,保证协程安全
counter++
}()
}
time.Sleep(1 * time.Second) // 等待一秒,等协程全部执行完(如果程序复杂,1s可能不够用)
t.Logf("counter = %d", counter)
}
// 输出结果如下:
=== RUN TestThreadSafe
channel_test.go:363: counter = 5000
// 结果正确,但是有一个问题。因为这里有个1秒的延迟等待,保证协程运行完毕再调用结果
--- PASS: TestThreadSafe (1.00s)
- 先来介绍下:同步等待组(WaitGroup)
waitGroup用于同步协程同步,等待一组协程执行完毕,才会继续向下执行.
1. 主协程调用Add()设置等待的协程数量.
2. 协程执行完毕,调用Done()函数
3. wait()函数阻塞,直到所有协程执行完毕才会继续向下执行.
- 方法2 : 使用同步等待队列(waitGroup)保证顺序执行
// 协程安全Demo
func TestWaitGroup(t *testing.T) {
var mut sync.Mutex // 互斥锁
var wg sync.WaitGroup // 等待队列
counter := 0
for i := 0; i < 5000; i++ {
wg.Add(1) // 加个任务
go func() {
defer func() {
mut.Unlock() //函数调用完成后:解锁,保证协程安全
}()
mut.Lock() // 函数将要调用前:加锁,保证协程安全
counter++
wg.Done() // 做完任务
}()
}
wg.Wait() //等待所有任务执行完毕
t.Logf("counter = %d", counter)
}
// 运行结果如下:
=== RUN TestWaitGroup
channel_test.go:382: counter = 5000
--- PASS: TestWaitGroup (0.00s)
channel的关闭和广播
- close 内置函数关闭一个channel,该通道必须是双向的或仅发送的
ch1 := make(chan int, 1)
ch2 := make(chan<- int, 1)
ch3 := make(<-chan int, 1)
close(ch1)
close(ch2)
close(ch3) // 报错 invalid operation: close(ch3) (cannot close receive-only channel)
- 向已经关闭的channel发送数据会panic
- 关闭一个已经关闭的channel会panic
- v, ok <- channel。 其中,ok为bool值,若ok == true时,表示channel处于open状态。 若ok==false时,表示channel处于close状态。
常见的并发场景
- 只执行一次(单例模式)
func TestOnceDo(t *testing.T) {
var once sync.Once
var wg sync.WaitGroup
for i := 0; i < 10; i++ {
wg.Add(1)
// 在多协程的情况下,保证某段代码只执行一次。
go func(ii int) {
once.Do(func() {
t.Log(ii)
})
wg.Done()
}(i)
}
wg.Wait()
}
// 输出结果
=== RUN TestOnceDo
channel_test.go:404: 0
--- PASS: TestOnceDo (0.00s)
- 发邮件,发短信
- 跑脚本
- 爬数据
总结
- 关键字
- go
- make
- chan
- select
- sync下的包
- 互斥锁
- 读写锁
- waitGroup
- map等
- 将复杂的任务拆分, 让goroutine去并发的执行,通过channel做数据通信
网友评论