一些并发的模型 - sheep44044的小站

1. 生产者-消费者模型#

这是最基础、最常用、也是最实用的模型。

有两个或多个生产者线程，它们生产数据（或任务）并放入一个共享的缓冲区。
有一个或多个消费者线程，它们从缓冲区中取出数据（或任务）并进行处理。
生产者和消费者之间通过这个缓冲区进行解耦。

核心挑战：

缓冲区为空：消费者不能从空缓冲区取数据。
缓冲区已满：生产者不能向已满的缓冲区放数据。
竞态条件：对缓冲区的访问必须是互斥的，防止多个线程同时修改导致数据不一致。

1
package main
2

3
import (
4
    "fmt"
5
    "time"
6
)
7

8
type Task struct {
9
    Runnable func(workerId int)
10
}
11

12
func main() {
13
    // 一个负责任务分发的管道
14
    ch := make(chan Task, 10)
15

16
    // 启动几个 worker 负责处理任务
17
    for id := range 10 {
18
        go func(workerId int) {
19
            for t := range ch {
20
                t.Runnable(workerId)
21
            }
22
        }(id)
23
    }
24

25
    // 任务分发
26
    for i := range 20 {
27
      j := i
28
        t1 := Task{
29
            Runnable: func(workerId int) {
30
                fmt.Printf("workerId%v：task%v做一件事情\n", workerId, j)
31
            },
32
        }
33
        ch <- t1
34
    }
35
    time.Sleep(1 * time.Second)
36
    close(ch)
37
}

现实应用：

消息队列（如Kafka, RabbitMQ）。
线程池任务调度。
Golang中的Channel，Java中的 BlockingQueue 其底层思想就是生产者-消费者模型。

2. 读者-写者问题#

核心思想：在读多写少的问题中，对读写锁的使用，写锁类似于互斥锁，读锁类似于共享锁（能开多个）。

读者只读取共享数据，不修改，写者会修改共享数据。允许多个读者同时读，但写者必须独占访问（即写时不能有读者或其他写者）。

核心挑战：

在保证数据一致性的前提下，最大限度地提高并发性（特别是对读者）。
可能导致写者饥饿：如果读者源源不断，写者可能永远无法获得访问权。

1
package main
2

3
import (
4
  "fmt"
5
  "sync"
6
  "time"
7
)
8

9
func main() {
10
  var (
11
    data    int
12
    rwMutex sync.RWMutex // 读写锁
13
    wg      sync.WaitGroup
14
  )
15

16
  // 写者
17
  wg.Add(1)
18
  go func() {
19
    defer wg.Done()
20
    for i := 0; i < 5; i++ {
21
      rwMutex.Lock() // 写锁
22
      data = i
23
      fmt.Printf("写入: %d\n", i)
24
      time.Sleep(time.Millisecond * 200)
25
      rwMutex.Unlock()
26
      time.Sleep(time.Millisecond * 100)
27
    }
28
  }()
29

30
  // 读者（多个）
31
  for i := 0; i < 3; i++ {
32
    wg.Add(1)
33
    go func(id int) {
34
      defer wg.Done()
35
      for j := 0; j < 5; j++ {
36
        rwMutex.RLock() // 读锁
37
        fmt.Printf("读者%d 读取: %d\n", id, data)
38
        time.Sleep(time.Millisecond * 50)
39
        rwMutex.RUnlock()
40
        time.Sleep(time.Millisecond * 100)
41
      }
42
    }(i)
43
  }
44

45
  wg.Wait()
46
}

现实应用：

数据库的并发控制。
文件系统、缓存的访问。

3. 哲学家就餐问题#

核心思想：避免多个进程（或线程）竞争有限资源时的死锁问题，通过特定的策略(破坏循环)或限制并发度（筷子数量）等方法，解决死锁问题。

五个哲学家围坐在一张圆桌旁，他们的一生只在做两件事：思考和吃饭。
桌上有五根筷子（不是五双），每两个哲学家之间有一根。
哲学家吃饭时需要同时拿起他左边和右边的筷子。
如果筷子在别人手里，他就必须等待。

核心挑战：

死锁：如果所有哲学家同时感到饥饿，并同时拿起自己左边的筷子，那么所有人都会永远等待右边的筷子被释放，导致系统僵住。

1
package main
2

3
import (
4
  "fmt"
5
  "sync"
6
  "time"
7
)
8

9
func main() {
10
  // 5根筷子（互斥锁）
11
  chopsticks := make([]sync.Mutex, 5)
12
  var wg sync.WaitGroup
13

14
  // 哲学家函数
15
  philosopher := func(id int) {
16
    defer wg.Done()
17

18
    left := id
19
    right := (id + 1) % 5
20

21
    // 避免死锁：编号为偶数的哲学家先拿左边，奇数先拿右边
22
    for i := 0; i < 3; i++ { // 吃3次
23
      if id%2 == 0 {
24
        chopsticks[left].Lock()
25
        chopsticks[right].Lock()
26
      } else {
27
        chopsticks[right].Lock()
28
        chopsticks[left].Lock()
29
      }
30

31
      // 拿到两根筷子，开始吃饭
32
      fmt.Printf("哲学家%d 正在吃饭...\n", id)
33
      time.Sleep(time.Millisecond * 100)
34

35
      // 放下筷子
36
      chopsticks[left].Unlock()
37
      chopsticks[right].Unlock()
38

39
      // 思考
40
      fmt.Printf("哲学家%d 正在思考...\n", id)
41
      time.Sleep(time.Millisecond * 100)
42
    }
43
  }
44

45
  // 启动5个哲学家
46
  for i := 0; i < 5; i++ {
47
    wg.Add(1)
48
    go philosopher(i)
49
  }
50

51
  wg.Wait()
52
}

现实应用：

任何需要获取多个互斥资源才能工作的场景，比如进程需要同时锁定多个数据库记录进行更新。

4. 理发师问题#

核心思想：模拟有容量限制的服务系统，通过空结构体(似乎叫信号量)来实现请求者与服务者的同步，并实现超容量时的拒绝和服务者的空闲/工作状态转换。

一个理发店有一个理发师（服务线程）、一把理发椅（临界资源）和N把供顾客等待的椅子（等待队列）。
如果没有顾客，理发师就在理发椅上睡觉。
顾客到来时，如果理发师在睡觉，就叫醒他理发。
如果理发师在忙，顾客就坐在等待椅上等待。
如果等待椅也坐满了，顾客就会离开。

核心挑战：

协调理发师（服务者）和顾客（请求者）的睡眠与唤醒。
管理有限的等待队列。

1
package main
2

3
import (
4
  "fmt"
5
  "math/rand"
6
  "sync"
7
  "time"
8
)
9

10
func main() {
11
  // 理发店参数
12
  const (
13
    waitingChairs = 3 // 等待椅数量
14
    totalCustomers = 10 // 总顾客数
15
  )
16

17
  var (
18
    waitingRoom = make(chan struct{}, waitingChairs) // 等待室（缓冲channel）
19
    wg          sync.WaitGroup
20
    barberReady = make(chan struct{}, 1) // 理发师就绪信号
21
  )
22

23
  // 理发师
24
  wg.Add(1)
25
  go func() {
26
    defer wg.Done()
27
    for {
28
      select {
29
      case <-waitingRoom:
30
        // 有顾客，开始理发
31
        barberReady <- struct{}{} // 理发师准备就绪
32
        fmt.Println("理发师正在理发...")
33
        time.Sleep(time.Millisecond * 200) // 理发时间
34
        fmt.Println("理发完成！")
35
      default:
36
        // 没有顾客，睡觉
37
        fmt.Println("理发师在睡觉...")
38
        time.Sleep(time.Millisecond * 500)
39
      }
40
    }
41
  }()
42

43
  // 顾客
44
  for i := 0; i < totalCustomers; i++ {
45
    wg.Add(1)
46
    go func(id int) {
47
      defer wg.Done()
48
      time.Sleep(time.Millisecond * time.Duration(rand.Intn(300)))
49

50
      select {
51
      case waitingRoom <- struct{}{}:
52
        // 成功进入等待室
53
        fmt.Printf("顾客%d 在等待理发\n", id)
54
        <-barberReady // 等待理发师就绪
55
        fmt.Printf("顾客%d 正在理发\n", id)
56
      default:
57
        // 等待室满了，离开
58
        fmt.Printf("顾客%d 看到等待室满了，离开了\n", id)
59
      }
60
    }(i)
61
  }
62

63
  // 等待所有顾客完成
64
  time.Sleep(time.Second * 3)
65
  fmt.Println("营业结束")
66
}

现实应用：

线程池或服务器模型。例如，一个Web服务器有固定数量的工作线程（理发师）和一个最大连接队列（等待椅）。当新请求到达时，如果有空闲线程则立即处理，否则进入队列等待，如果队列也满了，则返回“服务器繁忙”错误。