使用ticker定期定时加载内存中所有不断变化的路径中的文件

IT行业相对于一般传统行业，发展更新速度更快，一旦停止了学习，很快就会被行业所淘汰。所以我们需要踏踏实实的不断学习，精进自己的技术，尤其是初学者。今天golang学习网给大家整理了《使用ticker定期定时加载内存中所有不断变化的路径中的文件》，聊聊，我们一起来看看吧！

我有一个应用程序需要从两个不同的路径读取文件。读取所有这些文件后，我需要将它们加载到 products 地图的内存中。

路径：

• full：这是内存中服务器启动期间需要加载的所有文件的路径。该路径将包含大约 50 个文件，每个文件大小约为 60mb。
• delta：这是包含我们需要每 1 分钟定期加载到内存中的所有增量文件的路径。这些文件仅包含与完整路径文件的差异。该路径将包含大约 60 个文件，每个文件大小约为 20mb。

下面的代码 watchdeltapath 在服务器启动期间被调用以监视增量更改。它将从 getdeltapath 方法获取增量路径，并从该路径我需要加载内存中的所有文件。 此增量路径每隔几分钟就会发生变化，我不会错过任何一个增量路径以及该路径中的所有文件。

从 loadallfiles 方法加载内存中的所有文件可能需要一些时间（大约 5 分钟），所以我试图找到一种方法，让我不应该错过任何新的增量路径（因为它每隔几分钟就会不断变化）并且应该能够定期从增量路径加载内存中的所有这些文件，不会出现任何问题且高效。

我得到了下面的代码，每 1 分钟运行一次，每次都会查找新的 delta 路径 ，然后从内存中的该路径加载所有文件。它工作得很好，但我认为这不是正确的方法。如果 loadallfiles 方法需要超过 10 分钟来加载内存中的所有文件，并且我的代码每 1 分钟运行一次以查找新的增量路径，然后找到该新路径中的所有文件，然后加载到内存中，会发生什么情况？它会继续创建大量后台线程并可能大幅增加 cpu 使用率吗？

type applicationRepository struct {
  client         customer.Client
  logger         log.Logger
  done           chan struct{}
  products       *cmap.ConcurrentMap
}

// this will be called only once
func (r *applicationRepository) watchDeltaPath() error {
    ticker := time.NewTicker(1 * time.Minute)
    go func() {
        select {
        case <-r.done:
            ticker.Stop()
            return
        case <-ticker.C:
            func() (result error) {
                trans := r.logger.StartTransaction(nil, "delta-changes", "")
                defer trans.End()
                defer func() {
                    if result != nil {
                        trans.Errorf("Recovered from error: %v")
                    } else if err := recover(); err != nil {
                        trans.Errorf("Recovered from panic: %v", err)
                    }
                }()
                // get latest delta path everytime as it keeps changing every few minutes
                path, err := r.client.GetDeltaPath("delta")
                if err != nil {
                    return err
                }
                // load all the files in memory in "products" map from that path
                err = r.loadAllFiles(path)
                if err != nil {
                    return err
                }
                return nil
            }()
        }
    }()
    return nil
}

func (r *applicationRepository) Stop() {
    r.done <- struct{}{}
}

在产品中有效地做到这一点的最佳方法是什么？

这是我对代码的执行方式的玩弄 - https://go.dev/play/p/fs4-b0fwwte

正确答案

根据评论，“在产品中有效地做到这一点的最佳方法”取决于很多因素，并且在像 stack overflow 这样的网站上可能无法回答。话虽如此，我可以提出一种方法，可以让您更容易地思考如何最好地解决问题。

下面的代码（playground；非常粗糙且未经测试）演示了使用三个 go 例程的方法：

1. 检测新的增量路径并将其推送到缓冲通道
2. 处理初始负载
3. 等待初始加载完成，然后应用增量（请注意，这会处理初始加载正在进行时发现的增量）

如上所述，问题中没有足够的细节来确定这是否是一个好方法。初始负载和增量可能可以同时运行而不会使 io 饱和，但这需要测试（并且是一个相对较小的变化）。

// simulation of process to perform initial load and handle deltas
package main

import (
    "fmt"
    "strconv"
    "sync"
    "time"
)

const deltabuffer = 100
const initialloadtime = time.duration(time.duration(1.5 * float32(time.second)))
const deltacheckfrequency = time.duration(500 * time.millisecond)

func main() {
    ar := newapplicationrepository()
    time.sleep(5 * time.second)
    ar.stop()
    fmt.println(time.now(), "complete")
}

type applicationrepository struct {
    deltachan       chan string   // could be some other type...
    initialloaddone chan struct{} // closed when initial load finished

    done chan struct{}
    wg   sync.waitgroup
}

func newapplicationrepository() *applicationrepository {
    ar := applicationrepository{
        deltachan:       make(chan string, deltabuffer),
        initialloaddone: make(chan struct{}),
        done:            make(chan struct{}),
    }

    ar.wg.add(3)
    go ar.detectnewdeltas()
    go ar.initialload()
    go ar.deltaload()

    return &ar
}

// detectnewdeltas - watch for new delta paths
func (a *applicationrepository) detectnewdeltas() {
    defer a.wg.done()
    var previousdelta string
    for {
        select {
        case <-time.after(deltacheckfrequency):
            dp := a.getdeltapath()
            if dp != previousdelta {
                select {
                case a.deltachan <- dp:
                default:
                    panic("channel full - no idea what to do here!")
                }
                previousdelta = dp
            }
        case <-a.done:
            return
        }
    }
}

// getdeltapath in real application this will retrieve the delta path
func (a *applicationrepository) getdeltapath() string {
    return strconv.itoa(time.now().second()) // for now just return the current second..
}

// initialload - load the initial data
func (a *applicationrepository) initialload() {
    defer a.wg.done()
    defer close(a.initialloaddone)
    time.sleep(initialloadtime) // simulate time taken for initial load
}

// deltaload- load deltas found by detectnewdeltas
func (a *applicationrepository) deltaload() {
    defer a.wg.done()
    fmt.println(time.now(), "deltaload started")

    // wait for initial load to complete before doing anything
    <-a.initialloaddone
    fmt.println(time.now(), "initial load done")

    // wait for incoming deltas and load them
    for {
        select {
        case newdelta := <-a.deltachan:
            fmt.println(time.now(), newdelta)
        case <-a.done:
            return
        }
    }
}

// stop - signal loader to stop and wait until this is done
func (a *applicationrepository) stop() {
    close(a.done)
    a.wg.wait()
}

我想你想要 golang 并发模式：扇入、扇出。您可以在 google 中搜索。

我创建了一个示例代码。您可以复制粘贴它并创建文件夹 full 和 delta ，其中包含虚拟文件。

package main

import (
    "fmt"
    "log"
    "os"
    "path/filepath"
    "sync"
    "time"
)

type MyFile struct {
    full         map[string][]byte
    delta        map[string][]byte
    files        []string
    stopAutoLoad chan struct{}
}

func FilePathWalkDir(root string) ([]string, error) {
    var files []string
    err := filepath.Walk(root, func(path string, info os.FileInfo, err error) error {
        if !info.IsDir() {
            files = append(files, path)
        }
        return nil
    })
    return files, err
}

func main() {
    mf := NewMyFile()
    mf.StartAutoLoadDelta(10 * time.Second)

    // time.Sleep(15 * time.Second)
    // mf.StopAutoLoadDelta()

    time.Sleep(50 * time.Minute)
    fmt.Println(len(mf.full))
    fmt.Println(len(mf.delta))
}

func NewMyFile() *MyFile {
    mf := &MyFile{
        full:         make(map[string][]byte),
        delta:        make(map[string][]byte),
        stopAutoLoad: make(chan struct{}),
    }

    mf.LoadFile("full", 0)
    mf.LoadFile("delta", 0)
    return mf
}

func (mf *MyFile) StartAutoLoadDelta(d time.Duration) {
    ticker := time.NewTicker(d)

    go func() {
        defer func() {
            ticker.Stop()
        }()

        i := 1
        for {
            select {
            case <-ticker.C:
                // mf.deleteCurrentDelta()
                mf.LoadFile("delta", i)
                fmt.Println("In Memory:")
                for k, v := range mf.delta {
                    fmt.Printf("key : %s\t\tlen: %d\n", k, len(v))
                }
                i++
            case <-mf.stopAutoLoad:
                return
            }
        }
    }()
}

func (mf *MyFile) StopAutoLoadDelta() {
    fmt.Println("Stopping autoload Delta")
    mf.stopAutoLoad <- struct{}{}
}

func (mf *MyFile) deleteCurrentDelta() {
    for k, _ := range mf.delta {
        fmt.Println("data deleted: ", k)
        delete(mf.delta, k)
    }
}

type Fileinfo struct {
    name string
    data []byte
    err  error
}

func (mf *MyFile) LoadFile(prefix string, i int) {
    log.Printf("%s load : %d", prefix, i)
    files, err := FilePathWalkDir(prefix)
    if err != nil {
        panic("failed to open delta directory")
    }

    newFiles := make([]string, 0)

    for _, v := range files {
        if _, ok := mf.delta[v]; !ok {
            newFiles = append(newFiles, v)
        }
    }

    chanJobs := GenerateJobs(prefix, newFiles)
    chanResultJobs := ReadFiles(chanJobs, 8)
    counterTotal := 0
    counterSuccess := 0
    for results := range chanResultJobs {
        if results.err != nil {
            log.Printf("error creating file %s. stack trace: %s", results.name, results.err)
        } else {
            switch prefix {
            case "delta":
                mf.delta[results.name] = results.data
            case "full":
                mf.full[results.name] = results.data
            default:
                panic("not implemented")
            }
            counterSuccess++
        }
        counterTotal++
    }

    log.Printf("status jobs running: %d/%d", counterSuccess, counterTotal)
}

func GenerateJobs(prefix string, files []string) <-chan Fileinfo {
    chanOut := make(chan Fileinfo)

    go func() {
        for _, v := range files {
            chanOut <- Fileinfo{
                name: v,
            }
        }
        close(chanOut)
    }()

    return chanOut
}

func ReadFiles(chanIn <-chan Fileinfo, worker int) <-chan Fileinfo {
    chanOut := make(chan Fileinfo)

    var wg sync.WaitGroup

    wg.Add(worker)

    go func() {
        for i := 0; i < worker; i++ {
            go func(workerIndex int) {
                defer wg.Done()
                for job := range chanIn {
                    log.Printf("worker %d is reading file %s", workerIndex, job.name)
                    data, err := os.ReadFile(job.name)
                    chanOut <- Fileinfo{
                        name: job.name,
                        data: data,
                        err:  err,
                    }
                }
            }(i)
        }
    }()

    go func() {
        wg.Wait()
        close(chanOut)
    }()
    return chanOut
}

以上就是《使用ticker定期定时加载内存中所有不断变化的路径中的文件》的详细内容，更多关于的资料请关注golang学习网公众号！