为什么需要mutex

相比于 Go 语言宣扬的“用通讯的方式共享数据”，通过共享数据的方式来传递信息和协调线程运行的做法其实更加主流，毕竟大多数的现代编程语言，都是用后一种方式作为并发编程的解决方案的（这种方案的历史非常悠久，恐怕可以追溯到上个世纪多进程编程时代伊始了）

同步的用途有两个，一个是避免多个线程在同一时刻操作同一个数据块，另一个是协调多个线程，以避免它们在同一时刻执行同一个代码块。

使用互斥锁的注意事项

  • 不要重复锁定互斥锁
  • 不要忘记解锁互斥锁，必要时使用defer语句
  • 不要对尚未锁定或者已解锁的互斥锁解锁
  • 不要在多个函数之间直接传递互斥锁


读写锁

  • 对于某个受到读写锁保护的共享资源，多个写操作不能同时进行
  • 写操作和读操作也不能同时进行，但多个读操作却可以同时进行
  • 对写锁进行解锁，会唤醒“所有因试图锁定读锁，而被阻塞的 goroutine”
  • 对读锁进行解锁，只会在没有其他读锁锁定的前提下，唤醒“因试图锁定写锁，而被阻塞的 goroutine”；并且，最终只会有一个被唤醒的 goroutine 能够成功完成对写锁的锁定，其他的 goroutine 还要在原处继续等待。至于是哪一个 goroutine，那就要看谁的等待时间最长


  1. package main
  2. import (
  3.     "bytes"
  4.     "flag"
  5.     "fmt"
  6.     "io"
  7.     "io/ioutil"
  8.     "log"
  9.     "sync"
  10. )
  12. var protecting uint
  14. func init() {
  15.     flag.UintVar(&protecting, "protecting", 0, "It indicates whether to use a mutex to protect data writing.")
  16. }
  18. func main() {
  19.     flag.Parse()
  20.     var buffer bytes.Buffer
  22.     const (
  23.         max1 = 5 // 代表启动goroutine的数量
  24.         max2 = 10 // 代表每个goruntine需要写入的数据块的数量
  25.         max3 = 10 // 代表每个数据块有多少重复的数字
  26.     )
  28.     var mu sync.Mutex
  29.     sign := make(chan struct{}, max1)
  31.     for i := 1; i <= max1; i++ {
  32.         go func(id int, writer io.Writer) {
  33.             defer func() {
  34.                 sign <- struct{}{}
  35.             }()
  36.             for j := 1; j <= max2; j ++ {
  37.                 header := fmt.Sprintf("\n[id: %d, iteration: %d]", id , j)
  38.                 data := fmt.Sprintf(" %d", id * j)
  39.                 if protecting > 0 {
  40.                     mu.Lock()
  41.                 }
  42.                 _, err := writer.Write([]byte(header))
  43.                 if err != nil {
  44.                     log.Printf("error: %s [%d]", err, id)
  45.                 }
  46.                 for k := 0; k < max3; k++ {
  47.                     _, err := writer.Write([]byte(data))
  48.                     if err != nil {
  49.                         log.Printf("error: %s [%d]", err, id)
  50.                     }
  51.                 }
  52.                 if protecting > 0 {
  53.                     mu.Unlock()
  54.                 }
  55.             }
  56.         }(i, &buffer)
  57.     }
  59.     for i := 0; i < max1; i++ {
  60.         <- sign
  61.     }
  62.     data, err := ioutil.ReadAll(&buffer)
  63.     if err != nil {
  64.         log.Fatalf("fatal error: %s", err)
  65.     }
  66.     log.Printf("The contents:\n%s",data)
  67. }
  1. package main
  3. import (
  4.     "bytes"
  5.     "errors"
  6.     "fmt"
  7.     "io"
  8.     "log"
  9.     "sync"
  10.     "time"
  11. )
  13. // singleHandler 代表处理函数的类型
  14. type singleHandler func() (data string, n int, err error)
  16. // handlerConfig 代表处理流程配置的类型
  17. type handlerConfig struct {
  18.     handler singleHandler  // 单次处理函数
  19.     goNum int              // 需要启用的goroutine的数量
  20.     number int             // 单个goroutine中处理次数
  21.     interval time.Duration // 单个goroutine中处理的时间间隔
  22.     counter int            // 数据量计数器，以字节为单位
  23.     counterMu sync.Mutex   // 数据量计数器专用互斥锁
  24. }
  26. // count 会增加计数器的值，并会返回增加后的计数
  27. func (hc *handlerConfig) count(increment int) int {
  28.     hc.counterMu.Lock()
  29.     defer hc.counterMu.Unlock()
  30.     hc.counter += increment
  31.     return hc.counter
  32. }
  34. func main() {
  35.     // mu 代表以下流程要使用的互斥锁
  36.     var mu sync.Mutex
  38.     // genWriter 代表的是用于生成写入函数的函数
  39.     genWriter := func(writer io.Writer) singleHandler {
  40.         return func() (data string, n int, err error) {
  41.             // 准备数据
  42.             data = fmt.Sprintf("%s\t", time.Now().Format(time.StampNano))
  43.             // 写入数据
  44.             mu.Lock()
  45.             defer mu.Unlock()
  46.             n, err = writer.Write([]byte(data))
  47.             return
  48.         }
  49.     }
  51.     // genReader 代表的是用于生成读取函数的函数
  52.     genReader := func(reader io.Reader) singleHandler{
  53.         return func() (data string, n int, err error) {
  54.             buffer, ok := reader.(*bytes.Buffer)
  55.             if !ok {
  56.                 err = errors.New("unsupported reader")
  57.                 return
  58.             }
  59.             // 读取数据
  60.             mu.Lock()
  61.             defer mu.Unlock()
  62.             data, err = buffer.ReadString('\t')
  63.             n = len(data)
  64.             return
  65.         }
  66.     }
  68.     // buffer 代表缓冲区
  69.     var buffer bytes.Buffer
  71.     // 数据写入配置
  72.     writingConfig := handlerConfig {
  73.         handler: genWriter(&buffer),
  74.         goNum: 5,
  75.         number: 4,
  76.         interval: time.Millisecond * 100,
  77.     }
  79.     // sign 代表信号的通道
  80.     readingConfig := handlerConfig{
  81.         handler: genReader(&buffer),
  82.         goNum: 10,
  83.         number: 2,
  84.         interval: time.Millisecond * 100,
  85.     }
  87.     // sign 代表信号的通道
  88.     sign := make(chan struct{}, writingConfig.goNum + readingConfig.goNum)
  90.     // 启动多个goroutine对换重组进行多次数据写入
  91.     for i := 1; i <= writingConfig.goNum; i++ {
  92.         go func(i int) {
  93.             defer func() {
  94.                 sign <- struct{}{}
  95.             }()
  96.             for j := 1; j <= writingConfig.number; j ++ {
  97.                 time.Sleep(writingConfig.interval)
  98.                 data, n, err := writingConfig.handler()
  99.                 if err != nil {
  100.                     log.Printf("writer [%d-%d]: error: %s", i, j, err)
  101.                     continue
  102.                 }
  103.                 total := writingConfig.count(n)
  104.                 log.Printf("writer [%d-%d]: %s (total: %d)", i, j, data, total)
  105.             }
  106.         }(i)
  107.     }
  109.     // 启用多个goroutine对缓冲区进行多次数据读取
  110.     for i := 1; i <= readingConfig.goNum; i++ {
  111.         go func(i int){
  112.             defer func() {
  113.                 sign <- struct{}{}
  114.             }()
  115.             for j := 1; j<= readingConfig.number; j ++ {
  116.                 time.Sleep(readingConfig.interval)
  117.                 var data string
  118.                 var n int
  119.                 var err error
  120.                 for {
  121.                     data, n, err = readingConfig.handler()
  122.                     if err == nil || err != io.EOF {
  123.                         break
  124.                     }
  125.                     // 如果读比写快（读时会发生EOF错误），那就等一会再读
  126.                     time.Sleep(readingConfig.interval)
  127.                 }
  128.                 if err != nil {
  129.                     log.Printf("read [%d-%d]: error: %s", i, j, err)
  130.                     continue
  131.                 }
  132.                 total := readingConfig.count(n)
  133.                 log.Printf("reader [%d-%d]: %s (totl: %d)", i, j, data, total)
  134.             }
  135.         }(i)
  136.     }
  138.     // signNumber 代表需要接受信号的数量
  139.     signNumber := writingConfig.goNum + readingConfig.goNum
  140.     // 等待上面启用的所有goroutine的运行全部结束
  141.     for j := 0; j < signNumber; j ++ {
  142.         <-sign
  143.     }
  144. }