Goroutines异常行为修复
最后更新于：2022-04-02 06:51:45
在诸如守护进程这样的长期进程中，拥有一组长生命周期的goroutines非常普遍。这些goroutines通常被阻塞，等待被某种方式唤醒以继续工作。有时候，这些例程依赖于你没有很好控制的资源。也许一个goroutine会接收到Web服务中希望获取数据的请求，或者它正在监视一个临时文件。 如果程序处理不够健壮，goroutine会很容易陷入一个糟糕的状态。在长期运行的过程中，如果能创建一种机制来确保goroutine的健康状况良好，并在健康状况不佳时重新启动，那么我们的项目想必能活得久一点。 我们将在本节讨论对goroutines异常行为进行修复的话题。
我们将使用心跳来检查正在监测的goroutine的活跃程度。心跳的类型将取决于你想要监控的内容，但是如果你的goroutine可能会产生活锁，请确保心跳包含某种信息，以表明该goroutine不仅没死掉，而且还可以正常执行任务。在本节中，为了简单起见，我们只会考虑goroutines是活的还是死的。
下面这段代码建立一个管理者监视一个goroutine的健康状况，以及它的子例程。如果例程变得不健康，管理者将重新启动子例程。为此，它需要引用一个可以启动goroutine的函数。让我们看看管理程序是什么样子的：
```
type startGoroutineFn func(done <-chan interface{},
pulseInterval time.Duration) (heartbeat <-chan interface{}) //1
newSteward := func(timeout time.Duration, startGoroutine startGoroutineFn) startGoroutineFn { //2
return func(done <-chan interface{}, pulseInterval time.Duration) <-chan interface{} {
heartbeat := make(chan interface{})
go func() {
defer close(heartbeat)
var wardDone chan interface{}
var wardHeartbeat <-chan interface{}
startWard := func() { //3
wardDone = make(chan interface{}) //4
wardHeartbeat = startGoroutine(or(wardDone, done), timeout/2) //5
}
startWard()
pulse := time.Tick(pulseInterval)
monitorLoop:
for { //6
timeoutSignal := time.After(timeout)
for {
select {
case <-pulse:
select {
case heartbeat <- struct{}{}:
default:
}
case <-wardHeartbeat: //7
continue monitorLoop
case <-timeoutSignal: //8
log.Println("steward: ward unhealthy; restarting")
close(wardDone)
startWard()
continue monitorLoop
case <-done:
return
}
}
}
}()
return heartbeat
}
}
```
1. 这里我们定义一个可以监控和重新启动的goroutine的函数签名。 我们看到熟悉的done通道，以及熟悉的心跳模式写法。
2. 在这里我们设置了超时时间，并使用函数startGoroutine来启动它正在监控的goroutine。有趣的是，监控器本身返回一个startGoroutineFn，表示监控器自身也是可监控的。
3. 在这里我们定义一个闭包，它以同样的的方式来启动我们正在监视的goroutine。
4. 这是我们创建一个新通道，我们会将其传递给监控通道，以响应发出的停止信号。
5. 在这里，我们开启对目标goroutine的监控。如果监控器停止工作，或者监控器想要停止被监控区域，我们希望监控者也停止，因此我们将两个done通道都包含在逻辑中。我们传入的心跳间隔是超时时间的一半，但正如我们在“心跳”中讨论的那样，这可以调整。
6. 这是我们的内部循环，它确保监控者可以发出自己的心跳。
7. 在这里我们如果接收到监控者的心跳，就会知道它还处于正常工作状态，程序会继续监测循环。
8. 这里如果我们发现监控者超时，我们要求监控者停下来，并开始一个新的goroutine。然后开始新的监测。
我们的for循环有点杂乱，但如果你阅读过前面的章节，熟悉其中的模式，那么理解起来会相对简单。 接下来让我们试试看如果监控一个行为异常的goroutine，会发生什么：
```
log.SetOutput(os.Stdout)
log.SetFlags(log.Ltime | log.LUTC)
doWork := func(done <-chan interface{}, _ time.Duration) <-chan interface{} {
log.Println("ward: Hello, I'm irresponsible!")
go func() {
<-done // 1
log.Println("ward: I am halting.")
}()
return nil
}
doWorkWithSteward := newSteward(4*time.Second, doWork) // 2
done := make(chan interface{})
time.AfterFunc(9*time.Second, func() { // 3
log.Println("main: halting steward and ward.")
close(done)
})
for range doWorkWithSteward(done, 4*time.Second) { // 4
}
log.Println("Done")
```
1. 可以看到这个goroutine什么都没干，持续阻塞等待被取消，它同样不会发出任何表明自己正常信号。
2. 这里开始建立被监控的例程，其4秒后会超时。
3. 这里我们9秒后向done通道发出信号停止整个程序。
4. 最后，我们启动监控器并在其心跳范围内防止示例停止。
这会输出：
```
18:28:07 ward: Hello, I'm irresponsible!
18:28:11 steward: ward unhealthy; restarting 18:28:11 ward: Hello, I'm irresponsible!
18:28:11 ward: I am halting.
18:28:15 steward: ward unhealthy; restarting 18:28:15 ward: Hello, I'm irresponsible!
18:28:15 ward: I am halting.
18:28:16 main: halting steward and ward.
18:28:16 ward: I am halting.
18:28:16 Done
```
看起来工作正常。我们的监控器比较简单，除了取消操作和心跳所需信息之外不接收也不返回任何参数。我们可以用闭包强化一下：
```
doWorkFn := func(done <-chan interface{}, intList ...int) (startGoroutineFn, <-chan interface{}) {//1
intChanStream := make(chan (<-chan interface{}))//2
intStream := bridge(done, intChanStream)
doWork := func(done <-chan interface{}, pulseInterval time.Duration) <-chan interface{} {//3
intStream := make(chan interface{})//4
heartbeat := make(chan interface{})
go func() {
defer close(intStream)
select {
case intChanStream <- intStream://5
case <-done:
return
}
pulse := time.Tick(pulseInterval)
for {
valueLoop:
for _, intVal := range intList {
if intVal < 0 {
log.Printf("negative value: %v\n", intVal)//6
return
}
for {
select {
case <-pulse:
select {
case heartbeat <- struct{}{}: default:
}
case intStream <- intVal:
continue valueLoop
case <-done:
return
}
}
}
}
}()
return heartbeat
}
return doWork, intStream
}
```
1. 我们将监控器关闭的内容放入返回值，并返回所有监控器用来交流数据的通道。
2. 我们建立通道的通道，这是我们在前面章节中"bridge"模式的应用。
3. 这里我们建立闭包控制监控器的启动和关闭。
4. 这是各通道与监控器交互数据的实例。
5. 这里我们向起数据交互作用的通道传入数据。
6. 这里我们返回负数并从goroutine返回以模拟不正常的工作状态。
由于我们可能会启动监控器的多个副本，因此我们使用"bridge"模式来帮助向doWorkFn的调用者呈现单个不间断的通道。通过这样的方式，我们的监控器可以简单地通过组成模式而变得任意复杂。让我们看看如何调用：
```
log.SetFlags(log.Ltime | log.LUTC)
log.SetOutput(os.Stdout)
done := make(chan interface{})
defer close(done)
doWork, intStream := doWorkFn(done, 1, 2, -1, 3, 4, 5) //1
doWorkWithSteward := newSteward(1*time.Millisecond, doWork) //2
doWorkWithSteward(done, 1*time.Hour) //3
for intVal := range take(done, intStream, 6) { //4
fmt.Printf("Received: %v\n", intVal)
}
```
1. 这里我们调用该函数，它会将传入的不定长整数参数转换为可通信的流。
2. 在这里，我们创建了一个检查doWork关闭的监视器。我们预计这里会极快的进入失败流程，所以将监控时间设置为一毫秒。
3. 我们通知 steward 开启监测。
4. 最后，我们使用该管道，并从intStream中取出前六个值。
这会输出：
```
Received: 1
23:25:33 negative value: -1
Received: 2
23:25:33 steward: ward unhealthy; restarting Received: 1
23:25:33 negative value: -1
Received: 2
23:25:33 steward: ward unhealthy; restarting Received: 1
23:25:33 negative value: -1
Received: 2
```
我们可以看到监控器发现错误并重启。你可能还会注意到我们只接收到了1和2，这证明了重启功能正常。如果你的系统对重复值很敏感，一定要考虑对其进行处理。你也可以考虑在一定次数的失败后退出。比如在这样的位置：
```
valueLoop:
for _, intVal := range intList {
// ...
}
```
稍作修改：
```
valueLoop:
for {
intVal := intList[0]
intList = intList[1:]
// ...
}
```
尽管我们依然停留在返回的无效负数上，尽管我们的监控器将继续失败，但这会记录在重新启动前的位置，你可以在这个思路上扩展。
使用这样的方式可以确保你的系统保持健康，此外，相信系统崩溃的减少也能大幅度降低开发过程中猝死的几率。
愿诸君健康工作，准点下班。
* * * * *
学识浅薄，错误在所难免。我是长风，欢迎来Golang中国的群（211938256）就本书提出修改意见。
';