golang实现实时监听文件并自动切换目录

package watch
 
import (
	"bufio"
	"fmt"
	"github.com/fsnotify/fsnotify"
	"go.uber.org/zap"
	"io"
	"os"
	"path/filepath"
	"time"
)
 
// 存储已处理的位置
func saveProcessedOffset(fullFileName, logLevel string, offset int64) {
	logRunRecord := system.SysRunLogWatchRecord{}
	// 尝试从数据库中找到匹配的记录
	global.DB.Where(system.SysRunLogWatchRecord{FullFileName: fullFileName, LogLevel: logLevel}).First(&logRunRecord)
	// 如果找到了匹配的记录，则更新 offset 值
	if logRunRecord.ID > 0 {
		logRunRecord.ProcessedOffset = offset
		global.DB.Save(&logRunRecord)
	} else {
		// 没有找到匹配的记录，插入新记录
		newLogRecord := system.SysRunLogWatchRecord{
			FullFileName:    fullFileName,
			LogLevel:        logLevel,
			ProcessedOffset: offset,
		}
		global.DB.Create(&newLogRecord)
	}
}
 
// 从存储中读取已处理的位置
func readProcessedOffset(fullFileName, logLevel string) int64 {
	var logRunRecord system.SysRunLogWatchRecord
	global.DB.Where("full_file_name = ? and log_level = ?", fullFileName, logLevel).First(&logRunRecord)
	return logRunRecord.ProcessedOffset
}
 
// WatchSysRuntimeLogIncrement 检测日志
func WatchSysRuntimeLogIncrement(logPath, logLevel string) {
	watcher, err := fsnotify.NewWatcher()
	if err != nil {
		global.LOG.Error("fsnotify watch error:", zap.Error(err))
	}
	defer watcher.Close()
	ticker := time.NewTicker(1 * time.Minute)
	global.LOG.Info("启动一个定时器，每分钟检查一次时间并切换目录")
	defer ticker.Stop()
 
	directory := getCurrentLogDirectory(logPath)
	initCurrentErrorLog(fmt.Sprintf("%s/%s.log", directory, logLevel))
	err = watcher.Add(directory)
	if err != nil {
		global.LOG.Error("watcher Add error", zap.Error(err))
	}
	var currentReadFile *os.File
	// 在循环外打开文件
	logFilePath := fmt.Sprintf("%s/%s.log", directory, logLevel)
	// 当前正在解析的日志文件
	currentReadFile, err = os.Open(logFilePath)
	if err != nil {
		global.LOG.Error("无法打开文件:", zap.Error(err))
		return
	}
	var processedOffset int64
	for {
		select {
		case <-ticker.C:
			currentDate := time.Now().Format("2006-01-02")
			newDirectory := fmt.Sprintf("%s%s", logPath, currentDate)
			if newDirectory != directory {
				if _, err := os.Stat(fmt.Sprintf("%s/%s.log", newDirectory, logLevel)); err == nil {
					watcher.Remove(directory)
					directory = newDirectory
					err := watcher.Add(newDirectory)
					if err != nil {
						global.LOG.Error("无法监控新目录:", zap.Error(err))
					}
					global.LOG.Info("开始监控新文件:" + newDirectory)
					//监控新文件的时候，关闭旧文件
					currentReadFile.Close()
					logFilePath := fmt.Sprintf("%s/%s.log", directory, logLevel)
					currentReadFile, err = os.Open(logFilePath)
					processedOffset = 0
					global.LOG.Info("文件已经发生变化，新文件为：" + logFilePath)
				} else {
					global.LOG.Info("新文件不存在，继续监控旧文件")
					processedOffset = readProcessedOffset(fmt.Sprintf("%s/%s.log", directory, logLevel), logLevel)
				}
			} else {
				processedOffset = readProcessedOffset(fmt.Sprintf("%s/%s.log", directory, logLevel), logLevel)
			}
		case event, ok := <-watcher.Events:
			if !ok {
				return
			}
			if event.Op&fsnotify.Write == fsnotify.Write {
				global.LOG.Info("解析文件：" + logFilePath)
				if err != nil {
					global.LOG.Error("无法打开文件:", zap.Error(err))
					continue
				}
				currentReadFile.Seek(processedOffset, io.SeekStart)
				scanner := bufio.NewScanner(currentReadFile)
				for scanner.Scan() {
					line := scanner.Text()
					runLog := system.SystemRunningLog{
						Description: logLevel,
						ModuleName:  logPath,
						Operation:   line,
					}
					err = global.DB.Create(&runLog).Error
					if err != nil {
						global.LOG.Error("存储运行日志错误", zap.Error(err))
					}
				}
				if err := scanner.Err(); err != nil {
					global.LOG.Error("读取文件时发生错误", zap.Error(err))
				}
				// 更新已处理的位置
				processedOffset, err = currentReadFile.Seek(0, io.SeekEnd)
				if err != nil {
					global.LOG.Error("无法获取文件偏移量:", zap.Error(err))
				}
				// 将已处理的位置存储到文件中
				saveProcessedOffset(fmt.Sprintf("%s/%s.log", directory, logLevel), logLevel, processedOffset)
			}
		case err, ok := <-watcher.Errors:
			if !ok {
				return
			}
			global.LOG.Error("错误事件", zap.Error(err))
		}
	}
}
 
// 获取当前日期并构建日志目录路径
func getCurrentLogDirectory(logPath string) string {
	currentDate := time.Now().Format("2006-01-02")
	return fmt.Sprintf("%s%s", logPath, currentDate)
}
 
// 初始文件
func initCurrentErrorLog(errorLogPath string) {
	// 判断文件是否存在
	_, err := os.Stat(errorLogPath)
 
	if os.IsNotExist(err) {
		// 文件不存在，创建文件夹和文件
		err := os.MkdirAll(filepath.Dir(errorLogPath), os.ModePerm)
		if err != nil {
			global.LOG.Error("os.MkdirAll error:", zap.Error(err))
			return
		}
 
		file, err := os.Create(errorLogPath)
		if err != nil {
			global.LOG.Error("os.Create error:", zap.Error(err))
			return
		}
		defer file.Close()
 
		global.LOG.Info("error.log 文件已经存在,开始监控:" + errorLogPath)
	} else if err == nil {
		global.LOG.Info("error.log 文件已经存在,开始监控:" + errorLogPath)
	} else {
		global.LOG.Error("os.IsNotExist error:", zap.Error(err))
		return
	}
}