golang并发模型

• go在语言层面提供了内置的并发支持
• 不要通过共享内存来通信，而应该通过通信来共享内存

并发与并行

• 定义
  • 并发: 指同一时刻, 系统通过调度,来回切换交替的运行多个任务,看起来是"同时"进行的.一个处理器同时处理多个任务
  • 并行:指同一时刻,2个任务"真正的"同时进行.多个处理器或者多核的处理器同时处理多个不同的任务.
• 并行:多核cpu,物理上的同时执行.

image

• 并发:同一时刻,只能一条指令执行.通过快速轮换执行,宏观上是多个线程同时执行的.微观上不是同时执行的,只是把时间分成若干段,使得多个线程快速交替执行.(单核cpu,逻辑上的同时执行)

image

常见的并发编程模型

• 进程&线程(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

    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做数据通信