前情回顾

前文我们完成了日志采集系统基本功能，包括日志监控，日志采集，配置热更新，协程动态启动和关闭等。

本节目标

前文我们是通过将要监控的日志路径配置在配置文件中，根据配置文件热更新动态监控日志。
本节将etcd服务加入系统中，可以将要监控的日志文件路径和主题序列化为字符串保存到etcd中，这样系统可以监控etcd中该值得变化，从而动态启动协程和关闭协程监控指定日志。
这样可以通过etcd和配置文件两种方式监控日志。

图解系统监控日志流程

前情回顾

前文我们完成了日志采集系统的日志文件监控，配置文件热更新，协程异常检测和保活机制。

本节目标

本节加入kafka消息队列，kafka前文也介绍过了，可以对消息进行排队，解耦合和流量控制的作用，为什么一定要用kafka呢？主要原因就是在日志高并发读取后，如果直接将消息发给前端或者写入数据库，会造成崩溃或者卡死。kafka可以对消息进行排队和减轻压力，这样无论以后将这些消息录入数据库也好，传给前端分析也好，都能保证系统稳定性。代码我们也写过和测试了，只需要将之前写好的kafka读写消息代码整合过来即可。

主函数创建kafka生产者

在主函数中创建kafkaProducer,然后在defer中回收该资源。我们将该producer传递给每个监控日志的协程中，当日志有修改，就通过producer将修改的信息写入kafka，用kafka排队和缓存，可以提高稳定性，减少流量高峰。

前情回顾

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

本节目标

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

借尸还魂，增加异常处理

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

前情回顾

前文我们完成了如下目标
1 项目架构整体编写
2 使框架支持热更新

本节目标

在前文的框架基础上，我们
1 将之前实现的日志监控功能整合到框架中。
2 一个日志对应一个监控协程，当配置热更新后根据新配置动态关闭和启动协程。
3 编写测试代码，模拟向文件中不断写入日志，并备份日志，观察监控功能是否健壮。

前文中已经完成了文件的监控，kafka信息读写，今天主要完成配置文件的读写以及热更新。并且规划一下系统的整体结构，然后将之前的功能串起来形成一套完整的日志采集系统。

前情提要

上一节我们完成了如下目标
1 完成kafka消息读写
2 借助tailf实现文件监控，并模拟测试实时写文件以及文件备份时功能无误。

本节目标

1 编写系统结构，在主函数中加载配置
2 管理配置文件，实现热更新

日志统计系统的整体思路就是监控各个文件夹下的日志，实时获取日志写入内容并写入kafka队列，写入kafka队列可以在高并发时排队，而且达到了逻辑解耦合的目的。然后从kafka队列中读出数据，根据实际需求显示网页上或者控制台等。

前情提要

上一节我们完成了如下目标
1 配置kafka，并启动消息队列。
2 编写代码向kafka录入消息，并且从kafka读取消息。

本节目标

1 写代码从kafka中读取消息，保证kafka消息读写功能无误。
2 借助tailf实现文件监控，并模拟测试事实写文件以及文件备份时功能无误。
3 本系列文章开发语言使用Go

从kafka中读取消息

最近闹得沸沸扬扬的网易员工被强退事件，说点自己的看法。

事件始末

一个网易前员工，因为被检测出扩张型心肌病被网易提出单方面辞退，辞退原因是其绩效审核为D级，这位员工否认绩效审核结果，公司仍坚持辞退该员工，该员工要求索赔之后受到公司各种威胁以至于家人也受到牵连。具体事件在此处

项目结构

本系列文章意在记录如何搭建一个高可用的日志采集系统，实际项目中会有多个日志文件分布在服务器各个文件夹，这些日志记录了不同的功能。随着业务的增多，日志文件也再增多，企业中常常需要实现一个独立的日志采集系统，实时采集各个日志信息，并记录和输出到控制台或网页上，方便监控和查询。
本文日志采集系统架构如下

生产者消费者模型分析

操作系统中的经典模型，由若干个消费者和生产者，消费者消耗系统资源，生产者创造系统资源，资源的数量要保持在一个合理范围(小于数量上限，大约0)。而消费者和生产者是通过并发或并行方式访问系统资源的，需要保持资源的原子操作。
其实就是生产者线程增加资源数，如果资源数大于最大值则生产者线程挂起等待，当收到消费者线程的通知后继续生产。
消费者线程减少资源数，如果资源数为0，则消费者线程挂起，等待生产者通知后继续生产。
将该模型提炼成伪代码如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

func consume(){
    Lock()
    if count <= 0
        挂起等待(解锁，并等待资源数大于0)
        收到系统通知资源数大约0，抢占加锁
    count--
    如果当前资源数由最大值变少则通知生产者生产
    ULock()
}

func produce(){
    Lock()
    if count >= 最大值
        挂起等待(解锁，并等待资源数小于最大值)
        收到系统通知资源小于最大值，抢占加锁
    count++
    如果当前资源数由最小值0增加则通知消费者可以消耗
    ULock()
}

consume()消耗资源，produce()生产资源，之前实现过C版本的该模型
http://www.limerence2017.com/2017/08/08/pthreadwait/
C方式实现的是抢占式的，线程切换开销较大。下面给出golang协程方式的实现。

先实现资源的互斥访问

对于资源的互斥访问，其他语言提供了线程锁，golang也有线程锁，当然可以通过channel实现，这里我给出加锁访问资源的方式，因为channel内部也是通过加锁实现的，而且我习惯用channel做协程通信，对于共享资源的控制习惯用锁来控制，也比较高效。

非缓冲chan，读写对称

非缓冲channel,要求一端读取，一端写入。channel大小为零，所以读写操作一定要匹配。

1
2
3
4
5
6
7
8
9

func main() {
	nochan := make(chan int)
	go func(ch chan int) {
		data := <-ch
		fmt.Println("receive data ", data)
	}(nochan)
	nochan <- 5
	fmt.Println("send data ", 5)
}

我们启动了一个协程从channel中读取数据，在主协程中写入，程序的运行流程是主协程优先启动，运行到nochan<-5写入是阻塞，然后启动协程读取，从而完成协程间通信。
程序输出

1
2

receive data  5
send data  5