go 源码学习之---Tail 源码分析

已经有两个月没有写博客了,也有好几个月没有看go相关的内容了,由于工作原因最近在做java以及大数据相关的内容,导致最近工作较忙,博客停止了更新,正好想捡起之前go的东西,所以找了一个源码学习

这个也是之前用go写日志收集的时候用到的一个包 :github.com/hpcloud/tail, 这次就学习一下人家的源码,为了方便看这个代码,我将这个包进行了简化,也是用于方便理解,代码放到了:https://github.com/pythonsite/tail, 这个代码包可能无法正常用,只是为了方面理解tail这个包,以及学习人家的代码

精简后的代码目录

│  tail.go
│
└─watch
        filechanges.go
        inotify.go
        inotify_tracker.go
        watch.go

tail.go: 这里包含着tail包的核心代码,主要的逻辑处理时在这个里面

watch: 这个包主要用于对文件的监控,用于将文件的变化通知到tail.如:文件修改了,文件删除了,文件内容追加了

tail.go 代码分析

在tail.go中主要有几下几个结构体:

// Line 结构体用于存读每行的时候的对象
type Line struct {
    Text string            //当前行的内容
    Time time.Time        // 时间
    Err  error             // Error from tail
}

type SeekInfo struct {
    Offset         int64
    Whence        int    
}

// 关于配置的结构体
type Config struct {
    Location     *SeekInfo
    ReOpen        bool
    MustExist    bool        // 要打开的文件是否必须存在
    Poll        bool

    Pipe        bool

    Follow        bool        // 是否继续读取新的一行,可以理解为tail -f 命令
    
}

// 核心的结构体Tail
type Tail struct {
    Filename     string                    // 要打开的文件名
    Lines        chan *Line                // 用于存每行内容的Line结构体

    Config

    watcher     watch.FileWatcher
    changes     *watch.FileChanges

    tomb.Tomb    

    file         *os.File
    reader         *bufio.Reader
    lk sync.Mutex
}

Line 结构体用于存读取文件的每行内容

Tail 是核心的结构体,我们使用tail这个包的时候其实就是会先调用初始化这个struct的方法TailFile,如我在写日志收集的时候的使用:

    tail,err := tail.TailFile(conf.LogPath,tail.Config{
        ReOpen:true,
        Follow:true,
        Location:&tail.SeekInfo{Offset:0,Whence:2},
        MustExist:false,
        Poll:true,
    })

既然我们使用的时候就会在最开始的时候调用tail.TailFile方法,就直接看这个方法:

// 主要用于Tail结构体的初始化
func TailFile(filename string, config Config) (*Tail, error) {
    t := &Tail {
        Filename:    filename,
        Lines:        make(chan *Line),
        Config:        config,
    }
    t.watcher = watch.NewInotifyFileWatcher(filename)
    if t.MustExist {
        var err error
        t.file, err = OpenFile(t.Filename)
        if err != nil {
            return nil, err
        }
    }
    go t.tailFileSync()

    return t, nil
}

从这个代码里我们就可以看到它首先初始化了Tail结构体并且对Tail中的watcher进行的复制,先暂时不看watch相关的内容

然后就是关于文件是否必须存在的判断处理,最后开启了一个一个线程执行tailFileSync()方法,我们接着看tailFileSync方法

func (tail *Tail) tailFileSync(){
    defer tail.Done()
    defer tail.close()

    if !tail.MustExist {
        err := tail.reopen()
        if err != nil {
            if err != tomb.ErrDying {
                tail.Kill(err)
            }
            return
        }
    }

    tail.openReader()

    var offset int64
    var err error
    // 一行行读文件内容
    for {

        if !tail.Pipe {
            offset,err = tail.Tell()
            if err != nil {
                tail.Kill(err)
                return
            }
        }

        line, err := tail.readLine()
        if err == nil {
            // 将读取的一行内容放到chan中
            tail.sendLine(line)
        } else if err == io.EOF {
            // 表示读到文件的最后了
            // 如果Follow 设置为false的话就不会继续读文件
            if !tail.Follow {
                if line != "" {
                    tail.sendLine(line)
                }
                return
            }
            // 如果Follow设置为True则会继续读
            if tail.Follow && line != "" {
                err := tail.seekTo(SeekInfo{Offset: offset, Whence: 0})
                if err != nil {
                    tail.Kill(err)
                    return
                }
            }
            // 如果读到文件最后,文件并没有新的内容增加
            err := tail.waitForChanges()
            if err != nil {
                if err != ErrStop {
                    tail.Kill(err)
                }
                return
            }


        } else {
            // 既不是文件结尾,也没有error
            tail.Killf("error reading %s :%s", tail.Filename, err)
            return
        }


        select {
        case <- tail.Dying():
            if tail.Err() == errStopAtEOF {
                continue
            }
            return
        default:    
        }
    }
}

这个方法里主要是先调用了openReader方法,这个方法其实并没有做什么,只是对tail.reqader进行了赋值:tail.reader = bufio.NewReader(tail.file)

接着就是循环一行行的读文件

在循环里最开始判断了tail.Pipe的值,这个值一般开始我也并不会设置,所以默认就是false,所以就会执行tail.Tell()方法,这个方法主要是用于获取文件当前行的位置信息,下面是Tell的代码内容:

// 获取文件当前行的位置信息
func (tail *Tail) Tell()(offset int64, err error) {
    if tail.file == nil {
        return
    }
    offset, err = tail.file.Seek(0, os.SEEK_CUR)
    if err != nil {
        return
    }
    tail.lk.Lock()
    defer tail.lk.Unlock()
    if tail.reader == nil {
        return
    }
    offset -= int64(tail.reader.Buffered())
    return
}

接着会调用tail.readLine()方法,这个方法就是用于获取文件的一行内容,同时将一行内容实例化为Line对象,然后扔到管道tail.Lines中

//将读取的文件的每行内容存入到Line结构体中,并最终存入到tail.Lines的chan中
func (tail *Tail) sendLine(line string) bool {
    now := time.Now()
    lines := []string{line}

    for _, line := range lines {
        tail.Lines <- &Line {
            line,
            now,
            nil,
        }
    }
    return true
}

最后的大量if 判断其实主要是针对读到文件末尾后的一些操作,

Tail结构体在最后定义的时候有一个参数:Follow, 这个参数的目的就是当读到文件最后的时候是否继续读文件, 如果最开始设置了false,那么读到最后之后就不会在读文件了

如果设置为True,那么读到文件最后之后会保存文件的位置信息,并执行waitForChanges() 去等待文件的变化,waitForChanges()代码内容如下:

// 等待文件的变化事件
func (tail *Tail) waitForChanges() error {
    if tail.changes == nil {
        // 这里是获取文件指针的当前位置
        pos, err := tail.file.Seek(0,os.SEEK_CUR)
        if err != nil {
            return err
        }
        tail.changes, err = tail.watcher.ChangeEvents(&tail.Tomb, pos)
        if err != nil {
            return err
        }
    }
    // 和inotify中进行很巧妙的配合,这里通过select 来进行查看那个chan变化了,来知道文件的变化
    select {
    case <- tail.changes.Modified:            // 文件被修改
        return nil
    case <- tail.changes.Deleted:            // 文件被删除或者移动到其他目录
        tail.changes = nil
        // 如果文件被删除或者被移动到其他目录,则会尝试重新打开文件
        if tail.ReOpen {
            fmt.Printf("Re-opening moved/deleted file %s...",tail.Filename)
            if err := tail.reopen();err != nil {
                return err
            }
            fmt.Printf("Successfully reopened %s", tail.Filename)
            tail.openReader()
            return nil
        } else {
            fmt.Printf("Stoping tail as file not longer exists: %s", tail.Filename)
            return ErrStop
        }
    case <- tail.changes.Truncated:            // 文件被追加新的内容
        fmt.Printf("Re-opening truncated file %s....", tail.Filename)
        if err := tail.reopen();err != nil {
            return err
        }
        fmt.Printf("SuccessFuly reopend truncated %s", tail.Filename)
        tail.openReader()
        return nil
    case <- tail.Dying():
        return nil
    }
    panic("unreachable")
}

看到这里的时候其实就能感觉到,别人写的代码其实也并不是非常复杂,也是很普通的代码,但是你会觉得人家很多地方用的非常巧妙,

这段代码中主要的是的内容就是select部分,这个部分通过select监控

tail.changes.Modified

tail.changes.Deleted

tail.changes.Truncated

从而知道文件的变化,是修改了,还是删除了,还是追加内容了,这几个其实都是一个channel,这几个channel中的内容是怎么放进去的呢,接下来看watch包中的内容

watch包代码分析

首先先看一下watch包中的watch.go,这个里面其实就是定一个了一个FileWatcher的接口

type FileWatcher interface {
    BlockUntilExists(*tomb.Tomb) error

    ChangeEvents(*tomb.Tomb, int64) (*FileChanges, error)
}

接着我们看一下inotify.go文件,这个里面我们就可以看到定一个InotifyFileWatcher结构体,并且实现了FileWatcher 这个接口

type InotifyFileWatcher struct {
    Filename     string
    Size        int64
}

func NewInotifyFileWatcher(filename string) *InotifyFileWatcher {
    fw := &InotifyFileWatcher {
        filepath.Clean(filename),
        0,
    }
    return fw
}


// 关于文件改变事件的处理,当文件被修改了或者文件内容被追加了,进行通知
func (fw *InotifyFileWatcher) ChangeEvents(t *tomb.Tomb, pos int64) (*FileChanges, error) {
    err := Watch(fw.Filename)
    if err != nil {
        return nil, err
    }

    changes := NewFileChanges()
    fw.Size = pos

    go func() {
        events := Events(fw.Filename)

        for {
            prevSize := fw.Size

            var evt fsnotify.Event
            var ok bool
            
            select {
            case evt, ok = <- events:
                if !ok {
                    RemoveWatch(fw.Filename)
                    return
                }
            case <- t.Dying():
                RemoveWatch(fw.Filename)
                return
            }
            switch {
            case evt.Op & fsnotify.Remove == fsnotify.Remove:
                fallthrough
            case evt.Op & fsnotify.Rename == fsnotify.Rename:
                RemoveWatch(fw.Filename)
                changes.NotifyDeleted()
                return

            case evt.Op & fsnotify.Chmod == fsnotify.Chmod:
                fallthrough
            case evt.Op & fsnotify.Write == fsnotify.Write:
                fi, err := os.Stat(fw.Filename)
                if err != nil {
                    // 文件如果被删除了通知文件删除到chan
                    if os.IsNotExist(err) {
                        RemoveWatch(fw.Filename)
                        changes.NotifyDeleted()
                        return
                    }

                }
                fw.Size = fi.Size()

                if prevSize > 0 && prevSize > fw.Size {
                    // 表示文件内容增加了
                    changes.NotifyTruncated()
                } else {
                    // 表示文件被修改了
                    changes.NotifyModified()
                }

                prevSize = fw.Size
            }

        }
    }()
    return changes, nil
}

func (fw *InotifyFileWatcher) BlockUntilExists(t *tomb.Tomb) error {
    err := WatchCreate(fw.Filename)
    if err != nil {
        return err
    }
    defer RemoveWatchCreate(fw.Filename)
    if _, err := os.Stat(fw.Filename);!os.IsNotExist(err) {
        return err
    }
    events := Events(fw.Filename)
    for {
        select {
        case evt, ok := <- events:
            if !ok {
                return fmt.Errorf("inotify watcher has been closed")
            }
            evtName, err := filepath.Abs(evt.Name)
            if err != nil {
                return err
            }
            fwFilename, err := filepath.Abs(fw.Filename)
            if err != nil {
                return err
            }
            if evtName == fwFilename {
                return nil
            }
        case <- t.Dying():
            return tomb.ErrDying
        }
    }
    panic("unreachable")
}

实现的接口就两个方法:

ChangeEvents: 这个主要是监控文件的变化,是删除了,还是被修改了,或者是文件,然后将状态信息通过调用:changes.NotifyTruncated()或者

changes.NotifyDeleted() 或者changes.NotifyModified() 将状态信息更新到channel中,这样我们在分析tail.go 中最后的分析的那部分channel中的数据,就是在这里

放进去的

BlockUntilExists:这个主要是关于文件不存在的时候,如果最开始的时候可以允许文件不存在,那么就会 在这里通过for循环一直等待,知道文件存在

再看看filechanges.go 文件,代码内容如下:

type FileChanges struct {
    Modified  chan bool        // 修改
    Truncated chan bool        // 增加
    Deleted   chan bool        // 删除
}

func NewFileChanges() *FileChanges {
    return &FileChanges{
        make(chan bool, 1),
        make(chan bool, 1),
        make(chan bool, 1),
    }
}

func (fc *FileChanges) NotifyModified() {
    sendOnlyIfEmpty(fc.Modified)
}

func (fc *FileChanges) NotifyTruncated() {
    sendOnlyIfEmpty(fc.Truncated)
}

func (fc *FileChanges) NotifyDeleted() {
    sendOnlyIfEmpty(fc.Deleted)
}

func sendOnlyIfEmpty(ch chan bool) {
    select {
    case ch <- true:
    default:
    }
}

在这个里面也是可以学习到人家写的这个地方非常巧妙,虽然谈不上代码高达上,但是看着会让你很舒服,通过这个结构体,当文件被删除,修改和增加的时候就会让对应的channel中插入一个true,并且这里

的channel都是不带缓冲区的,只有当tail中触发一次之后,channel中的内容就会被获取出来,从而触发tail继续读文件的内容

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏木宛城主

SharePoint自动化部署,利用PowerShell 导入用户至AD——PART II

这是对上一篇文章《SharePoint自动化部署,利用PowerShell 导出/导入AD中的用户》进行补充。开发时,为了测试和演示,我们往往需要经常性的把用...

2458
来自专栏Java帮帮-微信公众号-技术文章全总结

Web-第十天 Cookie&Session学习

当用户访问某些Web应用时,经常会显示出该用户上一次的访问时间。例如,QQ登录成功后,会显示用户上次的登录时间。通过本任务,读者将学会如何使用Cookie技术实...

1513
来自专栏大内老A

WCF后续之旅(9):通过WCF的双向通信实现Session管理[上篇]

我们都知道,WCF支持Duplex的消息交换模式,它允许在service的执行过程中实现对client的回调。WCF这种双向通信的方式是我们可以以Event B...

1947
来自专栏木宛城主

基于Socket的网络聊天室编程(第一版)

一:什么是套接字 在网络编程中最常用的方案便是Client/Server (客户机/服务器)模型。在这种方案中客户应用程序向服务器程序请求服务。一个服务程序通常...

2665
来自专栏一个爱瞎折腾的程序猿

在asp.net core2.1中添加中间件以扩展Swashbuckle.AspNetCore3.0支持简单的文档访问权限控制

在此之前的接口项目中,若使用了 Swashbuckle.AspNetCore,都是控制其只在开发环境使用,不会就这样将其发布到生产环境(安全第一) 。 那么,...

1621
来自专栏恰同学骚年

设计模式的征途—19.命令(Command)模式

在生活中,我们装修新房的最后几道工序之一是安装插座和开关,通过开关可以控制一些电器的打开和关闭,例如电灯或换气扇。在购买开关时,用户并不知道它将来到底用于控制什...

562
来自专栏Greenplum

Linux 常用命令(三)

curl 命令支持在线下载功能,使用方便,它支持,S,等协议,还支持PUT,POST,COOKIES,认证授权等操作。

1770
来自专栏NetCore

Struts原理与实践

一、JDBC的工作原理 Struts在本质上是java程序,要在Struts应用程序中访问数据库,首先,必须搞清楚Java Database Connect...

2178
来自专栏智能大石头

NewLife.Redis基础教程

X组件缓存架构以ICache接口为核心,包括MemoryCache、Redis和DbCache实现,支持FX和netstandard2.0! 后续例程与使用说明...

1003
来自专栏程序员的碎碎念

JS动态加载以及JavaScript void(0)的爬虫解决方案

对于使用JS动态加载, 或者将下一页地址隐藏为 JavaScriptvoid(0)的网站, 如何爬取我们要的信息呢?

1352

扫码关注云+社区

领取腾讯云代金券