Go语言教程：并发处理与同步机制

本文将通过一个示例代码，深入浅出地介绍Go语言中的并发处理与同步机制。我们将逐步解析代码，帮助你理解其原理，并举一反三地应用到其他场景中。

示例代码概述#

该示例代码展示了如何使用Go语言的goroutine和通道（channel）来实现并发处理。代码中定义了一个数据处理配置结构体handlerConfig，并通过多个goroutine对缓冲区进行数据的并发写入和读取。

代码解析#

1. 导入必要的包#

1
import (
2
  "bytes"
3
  "errors"
4
  "fmt"
5
  "io"
6
  "log"
7
  "sync"
8
  "time"
9
)

这些包提供了我们需要的基本功能，如输入输出操作、错误处理、日志记录、并发控制等。

2. 定义处理函数类型和配置结构体#

1
type singleHandler func() (data string, n int, err error)
2

3
type handlerConfig struct {
4
  handler   singleHandler
5
  goNum     int
6
  number    int
7
  interval  time.Duration
8
  counter   int
9
  counterMu sync.Mutex
10
}

singleHandler：定义了一个处理函数类型，返回处理的数据、数据长度和错误信息。
handlerConfig：定义了处理配置，包括处理函数、goroutine数量、处理次数、间隔时间、计数器和计数器的互斥锁。

3. 计数器方法#

1
func (hc *handlerConfig) count(increment int) int {
2
  hc.counterMu.Lock()
3
  defer hc.counterMu.Unlock()
4
  hc.counter += increment
5
  return hc.counter
6
}

count方法用于安全地增加计数器的值，并返回增加后的计数。使用互斥锁counterMu来确保线程安全。

4. 主函数#

1
func main() {
2
  var mu sync.Mutex
3

4
  genWriter := func(writer io.Writer) singleHandler {
5
    return func() (data string, n int, err error) {
6
      data = fmt.Sprintf("%s\\t", time.Now().Format(time.StampNano))
7
      mu.Lock()
8
      defer mu.Unlock()
9
      n, err = writer.Write([]byte(data))
10
      return
11
    }
12
  }
13

14
  genReader := func(reader io.Reader) singleHandler {
15
    return func() (data string, n int, err error) {
16
      buffer, ok := reader.(*bytes.Buffer)
17
      if !ok {
18
        err = errors.New("unsupported reader")
19
        return
20
      }
21
      mu.Lock()
22
      defer mu.Unlock()
23
      data, err = buffer.ReadString('\\t')
24
      n = len(data)
25
      return
26
    }
27
  }
28

29
  var buffer bytes.Buffer
30

31
  writingConfig := handlerConfig{
32
    handler:  genWriter(&buffer),
33
    goNum:    5,
34
    number:   4,
35
    interval: time.Millisecond * 100,
36
  }
37

38
  readingConfig := handlerConfig{
39
    handler:  genReader(&buffer),
40
    goNum:    10,
41
    number:   2,
42
    interval: time.Millisecond * 100,
43
  }
44

45
  sign := make(chan struct{}, writingConfig.goNum+readingConfig.goNum)
46

47
  for i := 1; i <= writingConfig.goNum; i++ {
48
    go func(i int) {
49
      defer func() { sign <- struct{}{} }()
50
      for j := 1; j <= writingConfig.number; j++ {
51
        time.Sleep(writingConfig.interval)
52
        data, n, err := writingConfig.handler()
53
        if err != nil {
54
          log.Printf("writer [%d-%d]: error: %s", i, j, err)
55
          continue
56
        }
57
        total := writingConfig.count(n)
58
        log.Printf("writer [%d-%d]: %s (total: %d)", i, j, data, total)
59
      }
60
    }(i)
61
  }
62

63
  for i := 1; i <= readingConfig.goNum; i++ {
64
    go func(i int) {
65
      defer func() { sign <- struct{}{} }()
66
      for j := 1; j <= readingConfig.number; j++ {
67
        time.Sleep(readingConfig.interval)
68
        var data string
69
        var n int
70
        var err error
71
        for {
72
          data, n, err = readingConfig.handler()
73
          if err == nil || err != io.EOF {
74
            break
75
          }
76
          time.Sleep(readingConfig.interval)
77
        }
78
        if err != nil {
79
          log.Printf("reader [%d-%d]: error: %s", i, j, err)
80
          continue
81
        }
82
        total := readingConfig.count(n)
83
        log.Printf("reader [%d-%d]: %s (total: %d)", i, j, data, total)
84
      }
85
    }(i)
86
  }
87

88
  signNumber := writingConfig.goNum + readingConfig.goNum
89
  for j := 0; j < signNumber; j++ {
90
    <-sign
91
  }
92
}

5. 生成写入和读取函数#

genWriter：生成一个写入函数，向缓冲区写入当前时间戳。
genReader：生成一个读取函数，从缓冲区读取数据。

6. 配置写入和读取#

writingConfig：配置写入操作，包括处理函数、goroutine数量、处理次数和间隔时间。
readingConfig：配置读取操作。

7. 启动goroutine进行并发处理#

启动多个goroutine进行数据写入，每个goroutine按照配置的次数和间隔时间进行操作。
启动多个goroutine进行数据读取，处理读取过程中可能出现的EOF错误。

8. 等待所有goroutine完成#

使用通道sign来等待所有goroutine完成操作。

总结#

通过这个示例代码，我们学习了如何在Go语言中使用goroutine和通道来实现并发处理，并通过互斥锁确保数据的线程安全。你可以将这种模式应用到其他需要并发处理的场景中，如并发网络请求、并发文件处理等。

希望这个教程对你有所帮助！如果有任何问题，欢迎随时提问。