Go项目实战：打造高并发日志采集系统（五）

前情回顾

前文我们完成了如下功能
1 根据配置文件启动多个协程监控日志，并启动协程监听配置文件。
2 根据配置文件热更新，动态协调日志监控。
3 编写测试代码，向文件中不断写入日志并备份日志，验证系统健壮性。

本节目标

我们旨在编写一个健壮性较强的日志监控系统，不得不考虑这样一个问题，当某个日志监控协程崩溃或者异常退出，该如何处理？
我的想法是当监控日志文件的协程异常退出时，在主协程复活这个异常退出的协程，使其继续工作，这样极大的增强了系统的稳定性。

借尸还魂，增加异常处理

我们在tailf.go的WatchLogFile中增加异常处理，在协程崩溃时打印日志信息，并且向keychan中写入字符串通知主协程处理。

1
2
3
4
5
6
7
8
9
10
11
12
13
1func WatchLogFile(pathkey string, datapath string, ctx context.Context, keychan chan<- string) {

2   //省略之前写好的逻辑

3    //....

4   defer func() {

5       if errcover := recover(); errcover != nil {

6           fmt.Println("goroutine watch ", pathkey, " panic")

7           fmt.Println(errcover)

8           keychan <- pathkey

9       }

10    }()

11    //省略逻辑....

12}

13

WatchLogFile函数进行了扩充，增加了pathkey字符串表示监控哪个日志文件，pathkey实际是config.yaml中的路径的key值。
keychan 实际是一个缓冲chan，用来和主协程通信，告诉自己挂掉了。
WatchLogFile 中其余逻辑和前文一样，不做赘述。

主协程中增加复活逻辑

同样，在主协程中我们需要创建keychan这个缓冲chan，并且捕获子协程发过来的崩溃消息。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
1func main(){

2    //....省略

3    keyChan := make(chan string, KEYCHANSIZE)

4    //....

5    for {

6       select {

7       case pathData, ok := <-pathChan:

8           //省略...

9       case keystr := <-keyChan:

10          val, ok := configMgr[keystr]

11          if !ok {

12              continue

13          }

14          fmt.Println("recover goroutine watch ", keystr)

15          var ctxcover context.Context

16          ctxcover, val.ConfigCancel = context.WithCancel(context.Background())

17          go logtailf.WatchLogFile(keystr, val.ConfigValue,

18              ctxcover, keyChan)

19      }

20  }

21}

22

在主协程中愿有逻辑基础上，我们增加了keyChan的初始化，以及keychan数据的监听。
当主协程收到keychan的数据时，我们可以根据keystr修改其对应的context，并且启动新的协程继续监听该日志文件。
这样就达到了复活那些异常死掉的协程的目的。

keychan该如何回收

keychan被多个子协程引用，该如何回收？这种情况下多个子协程写数据，一个主协程接受数据，我们常用的策略如下：
1 不要在接受协程中关闭chan，因为如果此时有其他发送协程向关闭的chan写数据会导致崩溃。
2 有多个发送协程，等待最后一个发送协程退出时关闭chan可防止崩溃。
3 不作处理，等待系统回收，前提是保证所有协程正常退出，否则会导致资源泄漏。
我再三考虑了一下，统计最后协程退出会增加逻辑的复杂性，所以交给系统回收吧，但是我做好了协程的死锁检测和退出通知，应该不会有问题。

制造协程崩溃现场，模拟测试系统稳定性

为了测试我们的系统稳定性，我在修改WatchLogFile函数，新增如下处理，中断key值为logdir3的监控协程，进而观察主协程能否再次启动协程监听该日志文件。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
1func WatchLogFile(pathkey string, datapath string, ctx context.Context, keychan chan<- string) {

2   //...省略

3   defer func() {

4       if errcover := recover(); errcover != nil {

5           fmt.Println("goroutine watch ", pathkey, " panic")

6           fmt.Println(errcover)

7           keychan <- pathkey

8       }

9   }()

10

11  //模拟崩溃

12  if pathkey == "logdir3" {

13      panic("test panic ")

14  }

15  //...省略

16

然后我们启动日志系统，看到如下效果

从日志上可以看到每当我们的协程挂掉，主协程会启动新的协程继续监听日志，保证了系统的稳定性。
谢谢关注我的公众号
源码下载地址
https://github.com/secondtonone1/golang-/tree/master/logcatchsys