GOLANG中的并发

golang的一大特点是对并发的支持较好，golang的并发是通过goroutine来实现的。顾名思义，goroutine就是golang实现的协程。

我们说并发，可以是线程的并发，也可以是协程的并发，协程相对于线程的优点是协程比线程更轻量级，因此并发度可以更高。拿goroutine来说，一个go进程包含数千个goroutine同时在跑。

channel

golang中，每个goroutine是独立的，多个goroutine为了共同完成一个任务，需要有一定的通信机制。

比如说任务T由两个协程A、B共同完成，且A、B之间存在依赖关系，协程B依赖于协程A的执行结果，也就是只有等协程A执行完之后，协程B才能开始执行。

golang中协程之间的通信是通过channel来完成的。

可以把channel理解成一个管道（pipe），数据从管道的一端流进，从另一端流出。channel的语义是，当我们从管道中读数据时，读操作会一直block直到有数据流入管道；同样的，当我们写数据到管道中时，写操作一直block直到管道中的数据被读走。

unbuffered channel

unbuffered channel的语义是：当我们从管道中读数据时，读操作会一直block直到有数据流入管道；同样的，当我们写数据到管道中时，写操作一直block直到管道中的数据被读走。

下面的case我想在程序退出之前在屏幕上输出hello world，为了实现这点，我使用了done这个类型为chan bool的channel变量。

package main

import (
    "fmt"
    "time"
)
// 接收 bool 的 cahnnel
func hello(done chan bool) {
    fmt.Println("hello world")
    time.Sleep(4 * time.Second)
    done <- true
}
func main() {
  //用 make 建立一個不為 nil 的 channel
    done := make(chan bool)
    fmt.Println("Main going to call hello go goroutine")
    go hello(done)
    <- done // 管道读操作一直block，直到 hello goroutine执行并往管道中写数据，注释掉此行，main goroutine会一直执行到结束，hello goroutine不会被调度
    fmt.Println("Main received data")
}

复制代码

buffered channel

使用make创建channel的时候，除了类型之外，还可以指定另外一个参数capacity，capacity指定了channel的buffer长度，这种channel称之为buffered channel，capacity默认为0。

类似地，buffered channel的语义也很好理解：当buffer满了，继续写就会被block；当buffer空了，继续读就会被block。

package main

import (
    "fmt"
    "time"
)

func write(ch chan int) {
    for i := 0; i < 5; i++ {
        ch <- i
        fmt.Println("successfully wrote", i, "to ch")
    }
    close(ch)
}
func main() {
    ch := make(chan int, 2)
    go write(ch)
    time.Sleep(2 * time.Second)
    for v := range ch {
        fmt.Println("read value", v,"from ch")
        time.Sleep(2 * time.Second)

    }
}

/*
successfully wrote 0 to ch  
successfully wrote 1 to ch  
read value 0 from ch  
successfully wrote 2 to ch  
read value 1 from ch  
successfully wrote 3 to ch  
read value 2 from ch  
successfully wrote 4 to ch  
read value 3 from ch  
read value 4 from ch  
*/
复制代码

mutex

mutex实际上是一种锁机制，确保在任一时间点，只有一个goroutine能够进入到临界区（critical section），进而防止竞争条件（race condition）的发生。

下面的case中，启动1000个goroutine来让x自增1000次，每次运行的结果可能都不一定，x会小于等于1000。这是因为x的自增操作不是原子的，某一时刻，goroutine1读到x的值为10，+1之后为11，但是还没有写入主存，此时发生协程切换，goroutine2开始运行，goroutine2从主存读到x依然为10，+1之后为11，接下来，goroutine1和goroutine2把个字结果写回主存（不管先后顺序），x的值更新为11，出现了两次自增操作只实现了+1的效果。（local运行）

package main

import (
  "fmt"
  "sync"
)

var x = 0

func increment(wg *sync.WaitGroup, m *sync.Mutex) {
  x = x + 1
  wg.Done()
}
func main() {
  var w sync.WaitGroup
  for i := 0; i < 1000; i++ {
      w.Add(1)
      go increment(&w, &m) // 這裡一定要用 address
  }
  w.Wait()
  fmt.Println("final value of x", x)
}

复制代码

要解决这个问题很简单，每次执行自增操作之前先加锁，执行完之后再释放锁，以此来保证自增操作的原子性。下面的case不管运行多少次，每次运行x的值都会是1000，也就是说程序的运行结果是确定的。

package main

import (
  "fmt"
  "sync"
)

var x = 0

func increment(wg *sync.WaitGroup, m *sync.Mutex) {
  m.Lock()
  x = x + 1
  m.Unlock()
  wg.Done()
}
func main() {
  var w sync.WaitGroup
  var m sync.Mutex
  for i := 0; i < 1000; i++ {
      w.Add(1)
      go increment(&w, &m) // 這裡一定要用 address
  }
  w.Wait()
  fmt.Println("final value of x", x)
}

复制代码

特别的，我们还可以使用channel来实现mutex的功能。

package main

import (
  "fmt"
  "sync"
)

var x = 0

func increment(wg *sync.WaitGroup, m chan int) {
  m <- 1
  x = x + 1
  <- m
  wg.Done()
}
func main() {
  var w sync.WaitGroup
  m := make(chan int, 1)
  for i := 0; i < 1000; i++ {
      w.Add(1)
      go increment(&w, m) // 這裡一定要用 address
  }
  w.Wait()
  fmt.Println("final value of x", x)
}

复制代码

WaitGroup

除了channel和mutex之外，golang还提供了WaitGroup和Select来实现并发。

WaitGroup本质上是一个counter，只有counter=1的时候才会继续下一步。一般我们使用WaitGroup来实现语义：当一组goroutine都执行完成的时候，才继续下一步。

在下面的case中，执行一个goroutine之前counter先加1，goroutine执行完退出之前，counter减1，这就保证了只有在所有的goroutine都完成之后，才会继续执行main goroutine。

package main

import (
    "fmt"
    "sync"
    "time"
)

func process(i int, wg *sync.WaitGroup) {
    fmt.Println("started Goroutine ", i)
    time.Sleep(2 * time.Second)
    fmt.Printf("Goroutine %d ended\n", i)
    wg.Done() // -1
}

func main() {
    no := 3
    var wg sync.WaitGroup
    for i := 0; i < no; i++ {
        wg.Add(1) //  + 1
        go process(i, &wg) // wg 一定要用 pointer，否则每个 goroutine 都会有各自的 WaitGroup
    }
    wg.Wait() // =0 才继续下一步
    fmt.Println("All go routines finished executing")
}
复制代码

Select

select的语法跟switch的语法非常类似，用来实现如下语义：当一组协程中的所有协程都处于block时则block，当这组协程中有一个协程ready时，选择这个协程执行，当一组协程里面有多个协程ready时，随机选一个执行。

package main

import (
  "fmt"
  "time"
)

func server1(ch chan string) {
  time.Sleep(6 * time.Second)
  ch <- "from server1"
}
func server2(ch chan string) {
  time.Sleep(3 * time.Second)
  ch <- "from server2"

}
func main() {
  output1 := make(chan string)
  output2 := make(chan string)
  go server1(output1)
  go server2(output2)

  // 等待到其中一个 channel 回來，就执行，如果都有就会随机
  select {
  case s1 := <-output1:
      fmt.Println(s1)
  case s2 := <-output2:
      fmt.Println(s2)
  }
}
复制代码

总结以及下篇展望

本篇介绍了golang中关于并发编程的几个关键概念。golang的好处是从golang出发可以很清楚搞清楚并发中的很多关键概念。

下篇介绍并发编程中的关键概念，它们之间的联系，以及这些关键概念在golang中的实现。

critical section（临界区）、（race condition）竞争条件
同步原语、互斥变量、条件变量、信号量、锁

参考

本文代码示例均来自于：code examples