用于在工作流程中处理文件的 Go 例程。我缺少什么？

哈喽！今天心血来潮给大家带来了《用于在工作流程中处理文件的 Go 例程。我缺少什么？》，想必大家应该对Golang都不陌生吧，那么阅读本文就都不会很困难，以下内容主要涉及到，若是你正在学习Golang，千万别错过这篇文章~希望能帮助到你！

我正在构建一些东西来监视文件上传的目录。现在，我正在使用 for {} 循环连续读取目录以进行测试，并计划在将来使用 cron 或其他东西来启动我的应用程序。

目标是监视上传目录，确保文件已完成复制，然后将文件移动到另一个目录进行处理。文件本身的大小从 15gb 到大约 50gb 不等，我们每天会收到数百个文件。

这是我第一次尝试 go 例程。我不确定我是否完全误解了 go 例程、通道和等待组或其他东西，但我认为当我循环遍历文件列表时，每个文件都会由 go 例程函数独立处理。但是，当我运行下面的代码时，它会获取一个文件，但仅确认它在目录中找到的第一个文件。我注意到，一旦第一个文件完成，其他文件就会被确认为已完成。

package main

import (
        "flag"
        "fmt"
        "io/ioutil"
        "log"
        "os"
        "sync"
        "time"

        "gopkg.in/yaml.v2"
)

type Config struct {
        LogFileName  string `yaml:"logfilename"`
        LogFilePath  string `yaml:"logfilepath"`
        UploadRoot   string `yaml:"upload_root"`
        TPUploadTool string `yaml:"tp_upload_tool"`
}

var wg sync.WaitGroup

const WORKERS = 5

func getConfig(fileName string) (*Config, error) {
        conf := &Config{}
        yamlFile, err := os.Open(fileName)
        if err != nil {
                fmt.Printf("Error reading YAML file: %s\n", err)
                os.Exit(1)
        }

        defer yamlFile.Close()
        yaml_decoder := yaml.NewDecoder(yamlFile)

        if err := yaml_decoder.Decode(conf); err != nil {
                return nil, err
        }
        return conf, err
}

func getFileData(fileToUpload string, fileStatus chan string) {
        var newSize int64
        var currentSize int64

        currentSize = 0
        newSize = 0
        fmt.Printf("Uploading: %s\n", fileToUpload)
        fileDone := false
        for !fileDone {
                fileToUploadStat, _ := os.Stat(fileToUpload)
                currentSize = fileToUploadStat.Size()
                //fmt.Printf("%s current size is: %d\n", fileToUpload, currentSize)
                //fmt.Println("New size ", newSize)
                if currentSize != 0 {
                        if currentSize > newSize {
                                newSize = currentSize
                        } else if newSize == currentSize {
                                fileStatus <- "Done"
                                fileDone = true
                                wg.Done()
                        }
                }
                time.Sleep(1 * time.Second)
        }

}

func sendToCDS() {
        fmt.Println("Sending To CDS")
}

func main() {

        fileStatus := make(chan string)
        configFileName := flag.String("config", "", "YAML configuration file.\n")
        flag.Parse()

        if *configFileName == "" {
                flag.PrintDefaults()
                os.Exit(1)
        }

        UploaderConfig, err := getConfig(*configFileName)
        if err != nil {
                log.Fatal("Error reading configuration file.")
        }

        for {
                fmt.Print("Checking for new files..")
                uploadFiles, err := ioutil.ReadDir(UploaderConfig.UploadRoot)
                if err != nil {
                        log.Fatal(err)
                }
                if len(uploadFiles) == 0 {
                        fmt.Println("..no files to transfer.\n")
                }
                for _, uploadFile := range uploadFiles {
                        wg.Add(1)
                        fmt.Println("...adding", uploadFile.Name())
                        if err != nil {
                                log.Fatalln("Unable to read file information.")
                        }
                        ff := UploaderConfig.UploadRoot + "/" + uploadFile.Name()

                        go getFileData(ff, fileStatus)
                        status := <-fileStatus

                        if status == "Done" {
                                fmt.Printf("%s is done.\n", uploadFile.Name())
                                os.Remove(ff)
                        }
                }
                wg.Wait()
        }
}

我曾考虑过使用通道来实现线程安全的排队机制，该机制会加载目录中的文件，然后工作人员会拾取这些文件。我在 python 中做过类似的事情。

正确答案

由于以下行，代码按顺序处理每个文件：

go getfiledata(ff, filestatus)
status := <-filestatus

第一行创建一个 goroutine，但第二行等待该 goroutine 完成其工作。

如果您想并行处理文件，则可以使用工作池模式。

jobs:=make(chan string)
done:=make(chan struct{})
for i:=0;i<nworkers;i++ {
   go workerfunc(jobs,done)
}

jobs 通道将用于将新发现的文件发送给工作人员。当您发现新文件时，您可以简单地：

jobs <- filename

worker 应该处理该文件，并且应该返回从通道读取。所以它应该看起来像：

func worker(ch chan string,done chan struct{}) {
   defer func() {
      done<-struct{}{} // notify that this goroutine is completed
   }()
   for inputfile:=range ch {
       // process inputfile
   }
}

当一切完成后，您可以通过等待所有 goroutine 完成来终止程序：

close(jobs)
for i:=0;i<nWorkers;i++ {
   <-done
}

今天带大家了解了的相关知识，希望对你有所帮助；关于Golang的技术知识我们会一点点深入介绍，欢迎大家关注golang学习网公众号，一起学习编程~