消息传递模型

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
package main

import (
  "fmt"
  "time"
)

func main() {
  // 无缓冲区的channel (容量为0)
  ch1 := make(chan int)

  go func() {
    // 运行这一句时线程会阻塞, 只有当ch1被读取时, 42才被成功传递
    ch1 <- 42 
  }()

  fmt.Println("Received from ch1:", <-ch1)

  // 有缓冲区的channel, 缓冲区容量为2
  ch2 := make(chan string, 2)
  ch2 <- "hello"
  ch2 <- "world"
  // ch2 <- "!"  // 若执行则会阻塞

  fmt.Println("Received from ch2:", <-ch2)
  fmt.Println("Received from ch2:", <-ch2)

  // 关闭channel
  ch3 := make(chan int, 1)
  ch3 <- 100
  close(ch3)

  // 从已关闭的channel中收取数据, 当数据全部取出后, 再取则会得到默认值
  for v := range ch3 {
    fmt.Println("Received from closed ch3:", v)
  }

  // 取值时感知channel是否被关闭
  ch4 := make(chan int)
  go func() {
    time.Sleep(time.Second)
    close(ch4)
  }()

  v, ok := <-ch4
  if !ok {
    fmt.Println("ch4 is closed, no more data")
  } else {
    fmt.Println("Received from ch4:", v)
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
package main

import (
  "fmt"
  "time"
)

func main() {
  ch1 := make(chan string)
  ch2 := make(chan string)

  // 经过一定延迟再发送信息
  go func() {
    time.Sleep(1 * time.Second)
    ch1 <- "message from ch1"
  }()

  go func() {
    time.Sleep(2 * time.Second)
    ch2 <- "message from ch2"
  }()

  for i := 0; i < 2; i++ {
    // 并联ch1, ch2和timeout channel
    select {
    case msg1 := <-ch1:
      fmt.Println("Received:", msg1)
    case msg2 := <-ch2:
      fmt.Println("Received:", msg2)
    case <-time.After(3 * time.Second):
      {
        fmt.Println("Timeout: no message received")
        break
      }
    }
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
package main

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

// Worker模拟一个工作goroutine, 定期发送心跳. 
// 它会在一段时间后“死亡”, 停止发送心跳. 
func worker(
  ctx context.Context,
  id int,
  heartbeatInterval time.Duration,
  heartbeat chan<- int,
) {
  defer fmt.Printf("Worker %d 停止了\n", id)
  ticker := time.NewTicker(heartbeatInterval)
  defer ticker.Stop()

  // 模拟工作goroutine在一段时间后死亡
  deathTimer := time.NewTimer(5 * time.Second)
  defer deathTimer.Stop()

  for {
    select {
    case <-ctx.Done():
      return
    case <-deathTimer.C:
      // 工作goroutine“死亡”, 通过返回停止发送心跳
      fmt.Printf("Worker %d 死亡了\n", id)
      return
    case <-ticker.C:
      select {
      case heartbeat <- id:
        // 发送心跳
      case <-ctx.Done():
        return
      default: // 即使没有supervisor在监听心跳, 也不应该阻塞整个工作goroutine, 所有增加default
      }
    }
  }
}

// Supervisor监听工作goroutine的心跳, 如果某个工作goroutine停止发送心跳, 则重启它. 
func supervisor(
  ctx context.Context,
  workerCount int,
  heartbeatInterval time.Duration,
) {
  heartbeat := make(chan int)
  var wg sync.WaitGroup

  // 启动工作goroutine
  startWorker := func(id int) context.CancelFunc {
    workerCtx, cancel := context.WithCancel(ctx)
    wg.Add(1)
    go func() {
      defer wg.Done()
      worker(workerCtx, id, heartbeatInterval, heartbeat)
    }()
    return cancel
  }

  // 记录取消函数以便重启工作goroutine
  cancelFuncs := make(map[int]context.CancelFunc)
  for i := 1; i <= workerCount; i++ {
    cancelFuncs[i] = startWorker(i)
  }

  // 记录每个工作goroutine最后一次心跳时间
  lastHeartbeat := make(map[int]time.Time)
  for i := 1; i <= workerCount; i++ {
    lastHeartbeat[i] = time.Now()
  }

  checkInterval := heartbeatInterval * 2
  ticker := time.NewTicker(checkInterval)
  defer ticker.Stop()

  for {
    select {
    case <-ctx.Done():
      // 取消所有工作goroutine
      for _, cancel := range cancelFuncs {
        cancel()
      }
      wg.Wait()
      fmt.Println("Supervisor 停止了")
      return
    case id := <-heartbeat:
      // 更新最后一次心跳时间
      lastHeartbeat[id] = time.Now()
      fmt.Printf("Supervisor: 收到来自工作goroutine %d 的心跳\n", id)
    case <-ticker.C:
      // 检查哪些工作goroutine错过了心跳
      now := time.Now()
      for id, last := range lastHeartbeat {
        if now.Sub(last) > checkInterval {
          fmt.Printf("Supervisor: 工作goroutine %d 未收到心跳, 正在重启\n", id)
          // 取消旧工作goroutine并启动新工作goroutine
          cancelFuncs[id]() // 取消旧的
          cancelFuncs[id] = startWorker(id)
          lastHeartbeat[id] = time.Now() // 重置重启后心跳时间
        }
      }
    }
  }
}

func main() {
  ctx, cancel := context.WithCancel(context.Background())
  defer cancel()

  go supervisor(ctx, 3, 1*time.Second)

  // 让程序运行15秒
  time.Sleep(15 * time.Second)
  fmt.Println("主程序: 停止运行")
}