package fs import ( "ZhenTuLocalPassiveAdapter/config" "ZhenTuLocalPassiveAdapter/dto" "ZhenTuLocalPassiveAdapter/logger" "ZhenTuLocalPassiveAdapter/util" "context" "fmt" "path" "sort" "sync" "time" "github.com/aliyun/aliyun-oss-go-sdk/oss" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/codes" "go.uber.org/zap" ) var aliOssCache sync.Map type AliOSSAdapter struct { StorageConfig config.StorageConfig ossClient *oss.Client } func (a *AliOSSAdapter) getClient() (*oss.Client, error) { if a.ossClient == nil { client, err := oss.New( a.StorageConfig.AliOSS.Endpoint, a.StorageConfig.AliOSS.AccessKeyId, a.StorageConfig.AliOSS.AccessKeySecret, ) if err != nil { return nil, fmt.Errorf("创建阿里云OSS客户端失败: %w", err) } a.ossClient = client } return a.ossClient, nil } func (a *AliOSSAdapter) GetFileList(ctx context.Context, dirPath string, relDt time.Time) ([]dto.File, error) { _, span := tracer.Start(ctx, "GetFileList_alioss") defer span.End() span.SetAttributes(attribute.String("path", dirPath)) span.SetAttributes(attribute.String("relativeDate", relDt.Format("2006-01-02"))) if a.StorageConfig.AliOSS.Bucket == "" { span.SetAttributes(attribute.String("error", "未配置阿里云OSS存储桶")) span.SetStatus(codes.Error, "未配置阿里云OSS存储桶") return nil, fmt.Errorf("未配置阿里云OSS存储桶") } cacheKey := fmt.Sprintf("%s_%s", dirPath, relDt.Format("2006-01-02")) if cachedInterface, ok := aliOssCache.Load(cacheKey); ok { cachedItem := cachedInterface.(cacheItem) logger.Debug("缓存过期时间", zap.Duration("expiresIn", cachedItem.expires.Sub(time.Now()))) if time.Now().Before(cachedItem.expires) { logger.Debug("获取已缓存列表", zap.String("cacheKey", cacheKey)) span.SetAttributes(attribute.Bool("cache.hit", true)) return cachedItem.data, nil } } mutexKey := fmt.Sprintf("lock_%s", cacheKey) mutex, _ := aliOssCache.LoadOrStore(mutexKey, &sync.Mutex{}) lock := mutex.(*sync.Mutex) defer func() { // 解锁后删除锁(避免内存泄漏) aliOssCache.Delete(mutexKey) lock.Unlock() }() lock.Lock() if cachedInterface, ok := aliOssCache.Load(cacheKey); ok { cachedItem := cachedInterface.(cacheItem) logger.Debug("缓存过期时间", zap.Duration("expiresIn", cachedItem.expires.Sub(time.Now()))) if time.Now().Before(cachedItem.expires) { logger.Debug("过锁后获取已缓存列表", zap.String("cacheKey", cacheKey)) span.SetAttributes(attribute.Bool("aliOssCache.hit", true)) return cachedItem.data, nil } } client, err := a.getClient() if err != nil { span.SetAttributes(attribute.String("error", err.Error())) span.SetStatus(codes.Error, "创建阿里云OSS客户端失败") return nil, err } bucket, err := client.Bucket(a.StorageConfig.AliOSS.Bucket) if err != nil { span.SetAttributes(attribute.String("error", err.Error())) span.SetStatus(codes.Error, "获取存储桶失败") return nil, fmt.Errorf("获取存储桶失败: %w", err) } var fileList []dto.File prefix := path.Join(a.StorageConfig.AliOSS.Prefix, dirPath) marker := "" for { lsRes, err := bucket.ListObjects( oss.Prefix(prefix), oss.Marker(marker), oss.MaxKeys(1000), ) if err != nil { span.SetAttributes(attribute.String("error", err.Error())) span.SetStatus(codes.Error, "文件列表读取失败") return nil, fmt.Errorf("文件列表读取失败: %w", err) } for _, object := range lsRes.Objects { key := object.Key if !util.IsVideoFile(path.Base(key)) { continue } startTime, stopTime, err := util.ParseStartStopTime(path.Base(key), relDt) if err != nil { continue } if stopTime.IsZero() { stopTime = startTime } if startTime.Equal(stopTime) { stopTime = stopTime.Add(time.Second * time.Duration(config.Config.Record.Duration)) } // 生成预签名URL(有效期10分钟) signedURL, err := bucket.SignURL(key, oss.HTTPGet, 600) if err != nil { span.SetAttributes(attribute.String("error", err.Error())) span.SetStatus(codes.Error, "生成预签名URL失败") logger.Error("生成预签名URL失败", zap.Error(err)) continue } fileList = append(fileList, dto.File{ BasePath: a.StorageConfig.AliOSS.Bucket, Name: path.Base(key), Path: path.Dir(key), Url: signedURL, StartTime: startTime, EndTime: stopTime, }) } if !lsRes.IsTruncated { break } marker = lsRes.NextMarker } span.SetAttributes(attribute.Int("file.count", len(fileList))) sort.Slice(fileList, func(i, j int) bool { return fileList[i].StartTime.Before(fileList[j].StartTime) }) span.SetStatus(codes.Ok, "文件读取成功") cacheItem := cacheItem{ data: fileList, expires: time.Now().Add(30 * time.Second), } aliOssCache.Store(cacheKey, cacheItem) logger.Debug("缓存文件列表", zap.String("cacheKey", cacheKey)) return fileList, nil } // 添加定时清理缓存的初始化函数 func init() { go func() { ticker := time.NewTicker(1 * time.Minute) defer ticker.Stop() for { select { case <-ticker.C: cleanupAliOssCache() } } }() } // 添加缓存清理函数 func cleanupAliOssCache() { var keysToDelete []interface{} aliOssCache.Range(func(key, value interface{}) bool { item, ok := value.(cacheItem) if !ok { return true } if time.Now().After(item.expires) { keysToDelete = append(keysToDelete, key) } return true }) for _, key := range keysToDelete { aliOssCache.Delete(key) } }