machine-go/utils/common/task/task.go

package task

import (
	"bufio"
	"log"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"time"
)

type Task struct {
	Path           string
	FileMutex      *sync.Mutex
	TaskFunc       func(string, *Task)
	RuningTaskPath map[string]int
}

func NewTask(path string) *Task {

	var fileMutex sync.Mutex
	return &Task{
		Path:           path,
		FileMutex:      &fileMutex,
		TaskFunc:       nil,
		RuningTaskPath: make(map[string]int),
	}
}

func (task *Task) SetTaskFunc(f func(string, *Task)) {
	task.TaskFunc = f
}

func (task *Task) LogTask(msg string) error {
	task.FileMutex.Lock()
	defer task.FileMutex.Unlock()

	path := task.Path
	task.checkOrCreatePath(path)
	currentTime := time.Now().Unix()

	filePath := path + "/" + strconv.FormatInt(currentTime, 10) + ".log" // 目标文件路径

	file, err := os.OpenFile(filePath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
	if err != nil {
		log.Println("打开task文件失败:", err)
		return err
	}
	defer file.Close()

	// 写入内容到文件
	_, err = file.WriteString(msg + "\n")
	if err != nil {
		log.Println("写入task文件失败:", err)
		return err
	}

	return nil
}

func (task *Task) LogFailTask(msg string) error {
	task.FileMutex.Lock()
	defer task.FileMutex.Unlock()

	path := task.Path + "/fail/"
	task.checkOrCreatePath(path)

	currentTime := time.Now()
	currentHour := time.Date(currentTime.Year(), currentTime.Month(), currentTime.Day(), currentTime.Hour(), 0, 0, 0, currentTime.Location())
	timestamp := currentHour.Unix()

	filePath := path + strconv.FormatInt(timestamp, 10) + ".log" // 目标文件路径

	file, err := os.OpenFile(filePath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
	if err != nil {
		log.Println("打开task文件失败:", err)
		return err
	}
	defer file.Close()

	// 写入内容到文件
	_, err = file.WriteString(msg + "\n")
	if err != nil {
		log.Println("写入task文件失败:", err)
		return err
	}

	return nil
}

func (task *Task) checkOrCreatePath(path string) {
	dir := path

	// 检查任务目录是否存在
	_, err := os.Stat(dir)
	if os.IsNotExist(err) {
		// 任务目录不存在，创建任务目录
		err := os.MkdirAll(dir, 0755)
		if err != nil {
			log.Println(task.Path+" 目录创建失败:", err)
			return
		}

		log.Println(task.Path + " 目录创建成功")
	}
}

func (task *Task) TaskServer() {
	// 扫描接收文件协程
	go func() {
		task.ScanTask()
	}()
}

func (task *Task) ScanTask() {
	path := task.Path

	log.Println("扫描任务目录: " + path)
	for {
		currentTime := time.Now().Unix()

		dirRead, err := os.Open(path)

		if err != nil {
			time.Sleep(1 * time.Second)
			continue
		}
		for {
			files, err := dirRead.Readdir(10) // 每次读取10个文件
			if err != nil && len(files) == 0 {
				// 读取完所有文件或发生错误
				time.Sleep(500 * time.Microsecond)
				break
			}

			for _, file := range files {
				fileName := file.Name()
				ext := filepath.Ext(fileName)

				fileNameWithoutExt := strings.TrimSuffix(fileName, ext)

				fileTime, err := strconv.ParseInt(fileNameWithoutExt, 10, 64)
				if err != nil {
					continue // 跳过无法转换为时间戳的文件名
				}

				if ext == ".log" {
					if fileTime < currentTime-1 {
						filePath := path + "/" + fileName
						log.Println("扫描到文件: " + filePath)
						newFilePath := path + "/" + fileNameWithoutExt + ".scanned"
						err := os.Rename(filePath, newFilePath)
						if err != nil {
							log.Println(filePath + " 文件重命名失败: " + err.Error())
						}

						task.HandleTask(path + "/" + fileNameWithoutExt)
					}
				}

				if ext == ".scanned" {
					if fileTime < currentTime-600 {
						filePath := path + "/" + fileName
						// 非执行中的十分钟前未处理的重新处理
						if _, exists := task.RuningTaskPath[filePath]; !exists {
							log.Println("扫描到超时scanned需重新执行文件: " + filePath)
							task.HandleTask(path + "/" + fileNameWithoutExt)
						}
					}
				}
			}
		}

		dirRead.Close()
	}
}

func (task *Task) HandleTask(path string) {
	filePath := path + ".scanned"

	task.RuningTaskPath[path] = 1

	log.Println("开始处理文件: " + filePath)

	// 判断文件是否存在
	_, err := os.Stat(filePath)
	if os.IsNotExist(err) {
		log.Println(filePath + " 文件不存在")
		return
	}

	// 打开文件
	file, err := os.Open(filePath)

	if err != nil {
		log.Println(filePath + " 文件打开失败: " + err.Error())
		return
	}

	// 自定义缓冲区大小为 20 * 1MB
	const bufferSize = 20 * 1024 * 1024
	// 逐行读取文件内容
	scanner := bufio.NewScanner(file)
	// 设置缓冲区大小
	buf := make([]byte, bufferSize)
	scanner.Buffer(buf, bufferSize)

	i := 0 // 消息数计数器

	wg := sync.WaitGroup{}

	lineLimitChan := make(chan bool, 10) // 限制协程数量
	for scanner.Scan() {
		lineLimitChan <- true
		line := scanner.Text()
		if task.TaskFunc != nil {
			wg.Add(1)
			go func() {
				task.TaskFunc(line, task)
				<-lineLimitChan
				wg.Done()
			}()
		}
		i = i + 1
	}

	wg.Wait()

	if err := scanner.Err(); err != nil {
		err := os.Rename(filePath, path+".error")
		if err != nil {
			log.Println(filePath + " 文件重命名失败: " + err.Error())
		}
		log.Println(filePath + " 文件扫描出错: " + err.Error())
		return
	}

	// 文件扫描完毕
	file.Close()

	log.Printf(filePath+" 文件处理完毕, 共处理[%d]条信息", i)

	err = os.Remove(filePath)
	if err != nil {
		log.Println(filePath + " 文件删除失败: " + err.Error())
		return
	}

	log.Println(filePath + " 文件处理完成后删除")
	delete(task.RuningTaskPath, path)
}