feat: optimize caching strategy and implement concurrency limits

pkg/providers/cache.go

@@ -22,6 +22,8 @@ type ProviderCache struct {
 
 	cacheBlob      cache.Cache
 	cacheBlobLimit uint64
+	cacheSem       chan struct{}
+	backendSem     chan struct{}
 }
 
 func (c *ProviderCache) Close() error {
@@ -32,11 +34,21 @@ func NewProviderCache(
 	backend core.Backend,
 	cacheBlob cache.Cache,
 	cacheBlobLimit uint64,
+	cacheConcurrent uint64,
+	backendConcurrent uint64,
 ) *ProviderCache {
+	if cacheConcurrent == 0 {
+		cacheConcurrent = 16 // 默认限制 16 个并发缓存操作
+	}
+	if backendConcurrent == 0 {
+		backendConcurrent = 64 // 默认限制 64 个并发后端请求
+	}
 	return &ProviderCache{
 		parent:         backend,
 		cacheBlob:      cacheBlob,
 		cacheBlobLimit: cacheBlobLimit,
+		cacheSem:       make(chan struct{}, cacheConcurrent),
+		backendSem:     make(chan struct{}, backendConcurrent),
 	}
 }
 
@@ -81,6 +93,22 @@ func (c *ProviderCache) Open(ctx context.Context, owner, repo, id, path string,
 			Header:        respHeader,
 		}, nil
 	}
+
+	// 获取后端并发锁
+	select {
+	case c.backendSem <- struct{}{}:
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+
+	releaseBackend := func() { <-c.backendSem }
+	success := false
+	defer func() {
+		if !success {
+			releaseBackend()
+		}
+	}()
+
 	open, err := c.parent.Open(ctx, owner, repo, id, path, http.Header{})
 	if err != nil || open == nil {
 		if open != nil {
@@ -96,6 +124,14 @@ func (c *ProviderCache) Open(ctx context.Context, owner, repo, id, path string,
 		}
 		return nil, err
 	}
+
+	// 包装 Body 以在关闭时释放信号量
+	open.Body = &utils.CloserWrapper{
+		ReadCloser: open.Body,
+		OnClose:    releaseBackend,
+	}
+	success = true
+
 	if open.StatusCode == http.StatusNotFound {
 		// 缓存404路由
 		if err = c.cacheBlob.Put(ctx, key, map[string]string{
@@ -119,20 +155,34 @@ func (c *ProviderCache) Open(ctx context.Context, owner, repo, id, path string,
 		}
 		return open, nil
 	}
-	defer open.Body.Close()
-	allBytes, err := io.ReadAll(open.Body)
-	if err != nil {
-		return nil, err
+
+	// 尝试获取信号量进行缓存
+	select {
+	case c.cacheSem <- struct{}{}:
+		defer func() { <-c.cacheSem }()
+		defer open.Body.Close()
+		allBytes, err := io.ReadAll(open.Body)
+		if err != nil {
+			return nil, err
+		}
+		if err = c.cacheBlob.Put(ctx, key, map[string]string{
+			"Content-Length": open.Header.Get("Content-Length"),
+			"Last-Modified":  open.Header.Get("Last-Modified"),
+			"Content-Type":   open.Header.Get("Content-Type"),
+		}, bytes.NewBuffer(allBytes), time.Hour); err != nil {
+			zap.L().Warn("缓存归档失败", zap.Error(err), zap.Int("Size", len(allBytes)), zap.Uint64("MaxSize", c.cacheBlobLimit))
+		}
+		open.Body = utils.NopCloser{
+			ReadSeeker: bytes.NewReader(allBytes),
+		}
+		return open, nil
+	default:
+		// 无法获取信号量，直接流式返回，不进行缓存
+		zap.L().Debug("跳过缓存，并发限制已达", zap.String("path", path))
+		open.Body = &utils.SizeReadCloser{
+			ReadCloser: open.Body,
+			Size:       length,
+		}
+		return open, nil
 	}
-	if err = c.cacheBlob.Put(ctx, key, map[string]string{
-		"Content-Length": open.Header.Get("Content-Length"),
-		"Last-Modified":  open.Header.Get("Last-Modified"),
-		"Content-Type":   open.Header.Get("Content-Type"),
-	}, bytes.NewBuffer(allBytes), time.Hour); err != nil {
-		zap.L().Warn("缓存归档失败", zap.Error(err), zap.Int("Size", len(allBytes)), zap.Uint64("MaxSize", c.cacheBlobLimit))
-	}
-	open.Body = utils.NopCloser{
-		ReadSeeker: bytes.NewReader(allBytes),
-	}
-	return open, nil
 }