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繼續讀文件的內容

 

 


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM