使用ticker定期从经常变化的路径加载内存中的所有文件？

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

Path:

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

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

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

我得到下面的代码，每 1 分钟运行一次并寻找新的delta path每次然后加载内存中该路径的所有文件。它工作得很好，但我认为这不是正确的方法。如果发生什么情况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 https://go.dev/play/p/FS4-B0FWwTe

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

下面的代码（操场 https://go.dev/play/p/HLzu0Zauu91;相当粗糙且未经测试）演示了一种具有三个 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()
}