Errgroup 限流并行扇出

示例说明

扇出调用的策略抉择：一票否决还是容忍部分失败。聚合检索场景里单个分片过载不该废掉整个查询——任务函数吞掉错误记入失败列表并返回 nil，errgroup 不触发取消，其余分片照常完成；只有全部分片阵亡才判定整体失败。SetLimit 把瞬时并发压在阈值内，分片数再大也不会压垮下游。

技术要点

• SetLimit 上限内排队执行
• 部分失败语义聚合成功结果
• 全失败才返回整体错误

代码解读

fanout 的闭包里失败进 failures 切片、成功进 results，互斥锁保护两个切片的追加；闭包恒返回 nil 保证 errgroup 不取消兄弟任务。Wait 后按失败数判定：全败返回错误，否则返回成功的分片结果。示例的六分片里三号失败，五个成功结果照常返回。

使用建议

错误吞掉前要记录日志与指标。调用方需要知道部分失败时把 failures 一并返回。限流值与下游分片配额对齐。

代码示例

package main

import (
	"context"
	"errors"
	"fmt"
	"sync"

	"golang.org/x/sync/errgroup"
)

type ShardResult struct {
	ShardID int
	Rows    []string
}

func fetchShard(ctx context.Context, shard int) ([]string, error) {
	if shard == 3 {
		return nil, errors.New("shard 3 overloaded")
	}
	return []string{
		fmt.Sprintf("shard-%d-row-a", shard),
		fmt.Sprintf("shard-%d-row-b", shard),
	}, nil
}

func fanout(ctx context.Context, shards, concurrency int) ([]ShardResult, error) {
	g, ctx := errgroup.WithContext(ctx)
	g.SetLimit(concurrency)

	var mu sync.Mutex
	results := make([]ShardResult, 0, shards)
	var failures []error

	for shard := 0; shard < shards; shard++ {
		g.Go(func() error {
			rows, err := fetchShard(ctx, shard)
			if err != nil {
				mu.Lock()
				failures = append(failures, fmt.Errorf("shard %d: %w", shard, err))
				mu.Unlock()
				return nil // 单分片失败不取消全局，继续收其余结果
			}
			mu.Lock()
			results = append(results, ShardResult{ShardID: shard, Rows: rows})
			mu.Unlock()
			return nil
		})
	}

	_ = g.Wait()
	if len(failures) == shards {
		return nil, fmt.Errorf("all shards failed: %v", failures)
	}
	return results, nil
}

func main() {
	results, err := fanout(context.Background(), 6, 3)
	fmt.Println("shards succeeded:", len(results), "err:", err)
}