四、负载均衡、路由与集群

package loadbalance

import (
	"sync"
)

// ============================================================
// 轮询（Round Robin）负载均衡算法
// ============================================================

// RoundRobinPicker 是轮询负载均衡器
// 核心思想：维护一个计数器，每次请求时计数器 +1，
// 然后对实例总数取模，得到目标实例的索引
type RoundRobinPicker struct {
	mu        sync.Mutex    // 互斥锁，保证并发安全
	instances []string      // 可用实例列表，存储实例地址如 "172.10.0.1:9091"
	index     uint64        // 当前轮询到的位置
}

// NewRoundRobinPicker 创建一个轮询负载均衡器
// 参数 instances 是初始的可用实例列表
func NewRoundRobinPicker(instances []string) *RoundRobinPicker {
	return &RoundRobinPicker{
		instances: instances,
	}
}

// Pick 从可用实例中选择一个返回
// 每次调用返回下一个实例（按顺序循环）
func (r *RoundRobinPicker) Pick() string {
	r.mu.Lock()         // 加锁，防止并发时计数器混乱
	defer r.mu.Unlock() // 函数结束时释放锁

	if len(r.instances) == 0 {
		return "" // 没有可用实例，返回空字符串
	}

	// 取模运算得到当前索引
	// 例如有 3 个实例，index 依次为 0, 1, 2, 0, 1, 2, ...
	idx := r.index % uint64(len(r.instances))
	r.index++ // 计数器递增

	return r.instances[idx]
}

// Update 更新可用实例列表
// 当服务发现有变更时（新增或下线实例），调用此方法更新
func (r *RoundRobinPicker) Update(instances []string) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.instances = instances
}

package loadbalance

import (
	"sync"
)

// ============================================================
// 平滑加权轮询（Smooth Weighted Round Robin）负载均衡算法
// 参考 Nginx 的平滑加权轮询实现
// ============================================================

// WeightedInstance 代表一个带权重的实例
type WeightedInstance struct {
	Address           string // 实例地址，如 "172.10.0.1:9091"
	Weight            int    // 固定权重，用户配置的初始权重
	CurrentWeight     int    // 当前权重，算法运行过程中动态变化
	EffectiveWeight   int    // 有效权重，可根据调用结果动态调整
}

// SmoothWeightedRRPicker 是平滑加权轮询负载均衡器
type SmoothWeightedRRPicker struct {
	mu        sync.Mutex
	instances []*WeightedInstance // 所有可用实例
}

// NewSmoothWeightedRRPicker 创建一个平滑加权轮询负载均衡器
// 参数 instances 是带权重的实例列表
func NewSmoothWeightedRRPicker(instances []*WeightedInstance) *SmoothWeightedRRPicker {
	// 初始化：effectiveWeight 初始等于 weight
	for _, inst := range instances {
		inst.EffectiveWeight = inst.Weight
		inst.CurrentWeight = 0
	}
	return &SmoothWeightedRRPicker{
		instances: instances,
	}
}

// Pick 从可用实例中选择一个返回
// 核心算法：每次选择 currentWeight 最大的实例
func (p *SmoothWeightedRRPicker) Pick() string {
	p.mu.Lock()
	defer p.mu.Unlock()

	if len(p.instances) == 0 {
		return ""
	}

	// 第一步：计算 totalWeight（所有实例的 effectiveWeight 之和）
	// 同时更新每个实例的 currentWeight
	totalWeight := 0
	bestInstance := p.instances[0] // 记录 currentWeight 最大的实例

	for _, inst := range p.instances {
		// 更新当前权重：currentWeight += effectiveWeight
		inst.CurrentWeight += inst.EffectiveWeight
		// 累加总权重
		totalWeight += inst.EffectiveWeight

		// 找到 currentWeight 最大的实例
		if inst.CurrentWeight > bestInstance.CurrentWeight {
			bestInstance = inst
		}
	}

	// 第二步：被选中的实例，currentWeight 减去 totalWeight
	// 这一步是关键——它确保权重大的实例不会被连续选中
	// 减去 totalWeight 后，下次选择时该实例的 currentWeight 会大幅降低
	bestInstance.CurrentWeight -= totalWeight

	return bestInstance.Address
}

// OnSuccess 调用成功时调用，恢复有效权重
func (p *SmoothWeightedRRPicker) OnSuccess(addr string) {
	p.mu.Lock()
	defer p.mu.Unlock()
	for _, inst := range p.instances {
		if inst.Address == addr {
			// 调用成功，恢复有效权重（但不超过初始权重）
			if inst.EffectiveWeight < inst.Weight {
				inst.EffectiveWeight++
			}
			break
		}
	}
}

// OnError 调用失败时调用，降低有效权重
func (p *SmoothWeightedRRPicker) OnError(addr string) {
	p.mu.Lock()
	defer p.mu.Unlock()
	for _, inst := range p.instances {
		if inst.Address == addr {
			// 调用失败，降低有效权重
			// 这样下次选择时，该实例被选中的概率会降低
			inst.EffectiveWeight--
			// 注意：要防止 effectiveWeight 变成 0 或负数
			// 否则该实例将永远不被选中
			if inst.EffectiveWeight < 1 {
				inst.EffectiveWeight = 1
			}
			break
		}
	}
}

package loadbalance

import (
	"math/rand"
	"sync"
	"time"
)

// ============================================================
// 随机（Random）负载均衡算法
// ============================================================

// RandomPicker 是随机负载均衡器
type RandomPicker struct {
	mu        sync.Mutex
	instances []string
	rand      *rand.Rand // 随机数生成器
}

// NewRandomPicker 创建一个随机负载均衡器
func NewRandomPicker(instances []string) *RandomPicker {
	return &RandomPicker{
		instances: instances,
		// 使用当前时间作为随机种子，确保每次启动结果不同
		rand: rand.New(rand.NewSource(time.Now().UnixNano())),
	}
}

// Pick 随机选择一个实例
func (r *RandomPicker) Pick() string {
	r.mu.Lock()
	defer r.mu.Unlock()

	if len(r.instances) == 0 {
		return ""
	}

	// 生成一个 [0, len) 范围内的随机索引
	idx := r.rand.Intn(len(r.instances))
	return r.instances[idx]
}

package loadbalance

import (
	"math/rand"
	"sync"
	"time"
)

// ============================================================
// 加权随机（Weighted Random）负载均衡算法
// ============================================================

// WeightedRandomPicker 是加权随机负载均衡器
type WeightedRandomPicker struct {
	mu        sync.Mutex
	instances []*WeightedInstance
	rand      *rand.Rand
}

// NewWeightedRandomPicker 创建一个加权随机负载均衡器
func NewWeightedRandomPicker(instances []*WeightedInstance) *WeightedRandomPicker {
	return &WeightedRandomPicker{
		instances: instances,
		rand:      rand.New(rand.NewSource(time.Now().UnixNano())),
	}
}

// Pick 根据权重随机选择一个实例
// 原理：将权重看作区间长度，随机数落在哪个区间就选哪个实例
// 例如权重为 [5, 3, 2]，总权重为 10
// 实例1 占 [0, 5)，实例2 占 [5, 8)，实例3 占 [8, 10)
func (p *WeightedRandomPicker) Pick() string {
	p.mu.Lock()
	defer p.mu.Unlock()

	if len(p.instances) == 0 {
		return ""
	}

	// 第一步：计算总权重
	totalWeight := 0
	for _, inst := range p.instances {
		totalWeight += inst.Weight
	}

	// 第二步：生成一个 [0, totalWeight) 范围内的随机数
	r := p.rand.Intn(totalWeight)

	// 第三步：遍历实例，找到随机数落在的区间
	acc := 0 // 累加权重
	for _, inst := range p.instances {
		acc += inst.Weight
		if r < acc {
			return inst.Address
		}
	}

	// 理论上不会走到这里
	return p.instances[len(p.instances)-1].Address
}

package loadbalance

import (
	"hash/fnv"
	"sync"
)

// ============================================================
// 哈希（Hash）负载均衡算法
// ============================================================

// HashPicker 是哈希负载均衡器
// 核心思想：对请求的 key 计算 hash，然后对实例数取模
type HashPicker struct {
	mu        sync.Mutex
	instances []string
}

// NewHashPicker 创建一个哈希负载均衡器
func NewHashPicker(instances []string) *HashPicker {
	return &HashPicker{
		instances: instances,
	}
}

// Pick 根据 key 选择一个实例
// 参数 key 可以是用户 ID、请求 ID 等业务标识
// 相同的 key 总是路由到同一个实例（前提是实例列表不变）
func (p *HashPicker) Pick(key string) string {
	p.mu.Lock()
	defer p.mu.Unlock()

	if len(p.instances) == 0 {
		return ""
	}

	// 第一步：使用 FNV 哈希算法计算 key 的哈希值
	h := fnv.New32a()
	h.Write([]byte(key))
	hashValue := h.Sum32()

	// 第二步：对实例数量取模，得到目标实例索引
	idx := hashValue % uint32(len(p.instances))

	return p.instances[idx]
}

package loadbalance

import (
	"hash/fnv"
	"sort"
	"sync"
)

// ============================================================
// 一致性哈希（Consistent Hash）负载均衡算法
// ============================================================

// ConsistentHashPicker 是一致性哈希负载均衡器
// 核心思想：将节点和请求都映射到一个 0 ~ 2^32-1 的环上
// 请求顺时针找到的第一个节点就是目标节点
type ConsistentHashPicker struct {
	mu           sync.Mutex
	replicas     int            // 每个物理节点的虚拟节点数量
	ring         []uint32       // 哈希环，存储所有虚拟节点的哈希值（有序）
	hashMap      map[uint32]string // 哈希值 -> 物理节点地址的映射
}

// NewConsistentHashPicker 创建一个一致性哈希负载均衡器
// 参数 replicas 是每个物理节点的虚拟节点数，通常设为 100~200
// 虚拟节点越多，分布越均匀，但内存开销也越大
func NewConsistentHashPicker(replicas int) *ConsistentHashPicker {
	return &ConsistentHashPicker{
		replicas: replicas,
		hashMap:  make(map[uint32]string),
	}
}

// Add 添加物理节点到哈希环
// 每个物理节点会创建 replicas 个虚拟节点
func (p *ConsistentHashPicker) Add(nodes ...string) {
	p.mu.Lock()
	defer p.mu.Unlock()

	for _, node := range nodes {
		// 为每个物理节点创建 replicas 个虚拟节点
		for i := 0; i < p.replicas; i++ {
			// 虚拟节点的 key = 物理节点地址 + 编号
			// 例如 "172.10.0.1:9091#0", "172.10.0.1:9091#1", ...
			virtualKey := node + "#" + string(rune(i))
			hash := p.hashKey(virtualKey)

			// 将虚拟节点的哈希值加入环
			p.ring = append(p.ring, hash)
			// 记录哈希值对应的物理节点
			p.hashMap[hash] = node
		}
	}

	// 保持环有序，方便二分查找
	sort.Slice(p.ring, func(i, j int) bool {
		return p.ring[i] < p.ring[j]
	})
}

// Pick 根据 key 选择一个实例
// 原理：计算 key 的哈希值，在环上顺时针找到第一个节点
func (p *ConsistentHashPicker) Pick(key string) string {
	p.mu.Lock()
	defer p.mu.Unlock()

	if len(p.ring) == 0 {
		return ""
	}

	// 第一步：计算 key 的哈希值
	hash := p.hashKey(key)

	// 第二步：在环上二分查找第一个 >= hash 的虚拟节点
// sort.Search 返回第一个满足条件的索引
	idx := sort.Search(len(p.ring), func(i int) bool {
		return p.ring[i] >= hash
	})

	// 如果所有虚拟节点的哈希值都小于 hash，则回到环的起点（idx = 0）
	if idx == len(p.ring) {
		idx = 0
	}

	// 返回虚拟节点对应的物理节点地址
	return p.hashMap[p.ring[idx]]
}

// hashKey 计算字符串的哈希值
func (p *ConsistentHashPicker) hashKey(key string) uint32 {
	h := fnv.New32a()
	h.Write([]byte(key))
	return h.Sum32()
}

package loadbalance

import (
	"sync"
	"time"
)

// ============================================================
// 最少活跃数（Least Active）负载均衡算法
// ============================================================

// ActiveInstance 记录实例的活跃请求数
type ActiveInstance struct {
	Address      string
	ActiveCount  int64       // 当前活跃请求数（已发送但未收到响应）
	TotalTime    time.Duration // 总响应时间（用于计算平均响应时间）
	TotalCount   int64       // 总请求数
}

// LeastActivePicker 是最少活跃数负载均衡器
type LeastActivePicker struct {
	mu        sync.Mutex
	instances map[string]*ActiveInstance // 实例地址 -> 活跃信息
}

// NewLeastActivePicker 创建一个最少活跃数负载均衡器
func NewLeastActivePicker(addrs []string) *LeastActivePicker {
	m := make(map[string]*ActiveInstance)
	for _, addr := range addrs {
		m[addr] = &ActiveInstance{Address: addr}
	}
	return &LeastActivePicker{instances: m}
}

// Pick 选择活跃数最少的实例
func (p *LeastActivePicker) Pick() string {
	p.mu.Lock()
	defer p.mu.Unlock()

	var bestAddr string
	var minActive int64 = -1

	for addr, inst := range p.instances {
		// 找到活跃数最少的实例
		if minActive == -1 || inst.ActiveCount < minActive {
			minActive = inst.ActiveCount
			bestAddr = addr
		}
	}

	// 选中后，活跃数 +1
	if bestAddr != "" {
		p.instances[bestAddr].ActiveCount++
	}

	return bestAddr
}

// OnResponse 收到响应时调用
// 记录响应时间，活跃数 -1
func (p *LeastActivePicker) OnResponse(addr string, duration time.Duration) {
	p.mu.Lock()
	defer p.mu.Unlock()

	if inst, ok := p.instances[addr]; ok {
		inst.ActiveCount--  // 活跃数减一
		inst.TotalTime += duration
		inst.TotalCount++
	}
}

package loadbalance

import (
	"google.golang.org/grpc/balancer"
	"google.golang.org/grpc/resolver"
)

// ============================================================
// gRPC 负载均衡接口定义
// ============================================================

// PickerBuilder 接口：构建 Picker
// gRPC 在连接信息变更时会调用 Build 方法，传入最新的可用连接列表
// 我们需要返回一个 Picker 实例
type PickerBuilder interface {
	// Build 构建 Picker
	// 参数 readySCs 是所有就绪的子连接（SubConn），每个 SubConn 对应一个服务实例
	Build(readySCs map[balancer.SubConn]balancer.SubConnInfo) balancer.Picker
}

// Picker 接口：执行负载均衡
// 每次 RPC 调用时，gRPC 会调用 Pick 方法来选择一个连接
type Picker interface {
	// Pick 选择一个子连接
	// 参数 info 包含本次调用的上下文信息（如 metadata）
	// 返回选中的 SubConn 和是否完成
	Pick(info balancer.PickInfo) (balancer.PickResult, error)
}

package loadbalance

import (
	"sync"
	"sync/atomic"

	"google.golang.org/grpc/balancer"
	"google.golang.org/grpc/balancer/base"
	"google.golang.org/grpc/resolver"
)

// ============================================================
// gRPC 轮询负载均衡完整实现
// ============================================================

// roundRobinPicker gRPC 轮询 Picker 实现
type roundRobinPicker struct {
	subConns []balancer.SubConn // 所有可用的子连接
	mu       sync.Mutex
	index    uint64 // 当前轮询位置（使用原子操作保证并发安全）
}

// Pick 实现 Picker 接口
// 每次 RPC 调用时，gRPC 会调用此方法选择一个连接
func (p *roundRobinPicker) Pick(info balancer.PickInfo) (balancer.PickResult, error) {
	if len(p.subConns) == 0 {
		// 没有可用连接，返回错误
		return balancer.PickResult{}, balancer.ErrNoSubConnAvailable
	}

	// 原子递增索引，确保并发安全
	idx := atomic.AddUint64(&p.index, 1)
	// 取模得到当前连接的索引
	subConn := p.subConns[idx%uint64(len(p.subConns))]

	// 返回选中的连接
	// PickResult 包含 SubConn 和可选的 Done 回调函数
	return balancer.PickResult{
		SubConn: subConn,
	}, nil
}

// roundRobinPickerBuilder gRPC 轮询 PickerBuilder 实现
type roundRobinPickerBuilder struct{}

// Build 实现 PickerBuilder 接口
// gRPC 在连接列表变更时调用此方法，传入最新的可用连接
func (b *roundRobinPickerBuilder) Build(readySCs map[balancer.SubConn]balancer.SubConnInfo) balancer.Picker {
	// 将 map 转换为切片，方便轮询
	var subConns []balancer.SubConn
	for sc := range readySCs {
		subConns = append(subConns, sc)
	}

	return &roundRobinPicker{
		subConns: subConns,
	}
}

// ============================================================
// 注册自定义负载均衡器到 gRPC
// ============================================================

// BalancerName 负载均衡器名称
const BalancerName = "my_round_robin"

// RegisterBalancer 注册自定义负载均衡器
// 注册后，可以在创建 gRPC 客户端时通过 ServiceConfig 指定使用此负载均衡器
func RegisterBalancer() {
	// base.NewBalancerBuilder 创建一个基础负载均衡器
	// 第一个参数是负载均衡器名称
	// 第二个参数是 PickerBuilder
	// 第三个参数是配置选项
	base.NewBalancerBuilder(BalancerName, &roundRobinPickerBuilder{}, base.Config{})
}

// ============================================================
// 客户端使用示例
// ============================================================

// 客户端创建时指定负载均衡器的方式：
//
// import (
//     "google.golang.org/grpc"
//     "google.golang.org/grpc/balancer/roundrobin"
//     "google.golang.org/grpc/resolver"
// )
//
// func main() {
//     // 方式一：使用 gRPC 内置的 roundrobin
//     conn, err := grpc.Dial(
//         "etcd:///user-service",  // 使用 etcd 服务发现
//         grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy":"round_robin"}`),
//     )
//
//     // 方式二：使用自定义负载均衡器
//     RegisterBalancer() // 先注册
//     conn, err := grpc.Dial(
//         "etcd:///user-service",
//         grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy":"my_round_robin"}`),
//     )
// }

package loadbalance

import (
	"sync"

	"google.golang.org/grpc/balancer"
	"google.golang.org/grpc/balancer/base"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/resolver"
)

// ============================================================
// gRPC 平滑加权轮询负载均衡实现
// ============================================================

// weightedSubConn 包装 SubConn 和其权重信息
type weightedSubConn struct {
	subConn         balancer.SubConn // gRPC 子连接
	weight          int              // 固定权重
	currentWeight   int              // 当前权重（动态变化）
	effectiveWeight int              // 有效权重（可动态调整）
	addr            string           // 实例地址，用于从 metadata 获取权重
}

// smoothWeightedPicker gRPC 平滑加权轮询 Picker
type smoothWeightedPicker struct {
	mu       sync.Mutex
	subConns []*weightedSubConn
}

// Pick 实现 Picker 接口
func (p *smoothWeightedPicker) Pick(info balancer.PickInfo) (balancer.PickResult, error) {
	if len(p.subConns) == 0 {
		return balancer.PickResult{}, balancer.ErrNoSubConnAvailable
	}

	p.mu.Lock()
	defer p.mu.Unlock()

	// 第一步：计算总权重，更新每个实例的 currentWeight
	totalWeight := 0
	best := p.subConns[0]

	for _, sc := range p.subConns {
		sc.currentWeight += sc.effectiveWeight
		totalWeight += sc.effectiveWeight

		if sc.currentWeight > best.currentWeight {
			best = sc
		}
	}

	// 第二步：被选中的实例 currentWeight 减去 totalWeight
	best.currentWeight -= totalWeight

	// 第三步：返回结果，设置 Done 回调用于动态调整权重
	return balancer.PickResult{
		SubConn: best.subConn,
		Done: func(info balancer.PickDoneInfo) {
			p.mu.Lock()
			defer p.mu.Unlock()

			if info.Err != nil {
				// 调用失败，降低有效权重
				best.effectiveWeight--
				if best.effectiveWeight < 1 {
					best.effectiveWeight = 1 // 防止降到 0
				}
			} else {
				// 调用成功，恢复有效权重
				if best.effectiveWeight < best.weight {
					best.effectiveWeight++
				}
			}
		},
	}, nil
}

// smoothWeightedBuilder gRPC 平滑加权轮询 Builder
type smoothWeightedBuilder struct{}

// Build 实现 PickerBuilder 接口
// 从 SubConnInfo 中提取权重信息
func (b *smoothWeightedBuilder) Build(readySCs map[balancer.SubConn]balancer.SubConnInfo) balancer.Picker {
	var subConns []*weightedSubConn

	for sc, info := range readySCs {
		// 从 resolver.Address 的 Attributes 中获取权重
		// 权重是在服务发现阶段写入的
		weight := getWeightFromAddress(info.Address)
		if weight < 1 {
			weight = 1 // 默认权重为 1
		}

		subConns = append(subConns, &weightedSubConn{
			subConn:         sc,
			weight:          weight,
			effectiveWeight: weight,
			currentWeight:   0,
			addr:            info.Address.Addr,
		})
	}

	return &smoothWeightedPicker{subConns: subConns}
}

// getWeightFromAddress 从 resolver.Address 中读取权重
// 权重数据是在服务发现阶段写入 Attributes 的
func getWeightFromAddress(addr resolver.Address) int {
	// 这里简化了从 Attributes 获取权重的过程
	// 实际实现中，权重是通过 addr.Attributes 获取的
	// 通常在服务发现的 Resolver 中将权重写入
	return 1 // 默认返回 1
}

// 注册负载均衡器
const SmoothWeightedBalancerName = "smooth_weighted_rr"

func RegisterSmoothWeightedBalancer() {
	base.NewBalancerBuilder(SmoothWeightedBalancerName, &smoothWeightedBuilder{}, base.Config{})
}

package registry

// ============================================================
// 服务实例增加权重字段
// ============================================================

// ServiceInstance 增加了 Weight 字段
type ServiceInstance struct {
	ServiceName string            `json:"serviceName"`
	InstanceID  string            `json:"instanceId"`
	Address     string            `json:"address"`
	Port        int               `json:"port"`
	Metadata    map[string]string `json:"metadata"`
	Weight      int               `json:"weight"` // 新增：权重字段
}

// 在服务发现阶段，将权重写入 resolver.Address 的 Attributes
// 这样负载均衡器就可以从 Attributes 中读取权重了

// 示例：在 etcd Resolver 中写入权重
/*
func (r *etcdResolver) updateClientConn(instances []*registry.ServiceInstance) {
	var addrs []resolver.Address
	for _, inst := range instances {
		addr := resolver.Address{
			Addr: fmt.Sprintf("%s:%d", inst.Address, inst.Port),
			// 将权重写入 Attributes
			Attributes: attributes.New("weight", inst.Weight),
		}
		addrs = append(addrs, addr)
	}
	r.cc.UpdateState(resolver.State{
		Addresses: addrs,
	})
}
*/

package loadbalance

import (
	"context"
	"sync"

	"google.golang.org/grpc/balancer"
	"google.golang.org/grpc/balancer/base"
	"google.golang.org/grpc/metadata"
)

// ============================================================
// 分组路由实现：支持 A/B 流量分发
// ============================================================

// 分组相关的 context key
type groupKey struct{}

// WithGroup 将分组信息写入 context
// 调用方使用此函数设置当前请求的分组
// 例如：ctx = WithGroup(ctx, "A") 表示此请求属于 A 组
func WithGroup(ctx context.Context, group string) context.Context {
	return context.WithValue(ctx, groupKey{}, group)
}

// GetGroup 从 context 中读取分组信息
// 负载均衡器使用此函数获取当前请求应该发到哪个组
func GetGroup(ctx context.Context) string {
	if v, ok := ctx.Value(groupKey{}).(string); ok {
		return v
	}
	return "" // 没有设置分组，返回空字符串
}

// groupedSubConn 带分组信息的子连接
type groupedSubConn struct {
	subConn balancer.SubConn
	group   string // 该实例所属的分组，如 "A" 或 "B"
}

// groupPicker 分组路由 + 负载均衡 Picker
// 先按分组过滤，再在组内做轮询
type groupPicker struct {
	mu              sync.Mutex
	groupedSubConns map[string][]balancer.SubConn // 按分组组织的连接列表
	index           uint64                         // 轮询计数器
}

// Pick 实现 Picker 接口
func (p *groupPicker) Pick(info balancer.PickInfo) (balancer.PickResult, error) {
	// 第一步：从 context 中获取请求的分组标记
	group := GetGroup(info.Ctx)

	p.mu.Lock()
	defer p.mu.Unlock()

	// 第二步：根据分组筛选可用的连接
	// 如果指定了分组，只从该分组的连接中选择
	// 如果没有指定分组，从所有连接中选择
	var candidates []balancer.SubConn
	if group != "" {
		candidates = p.groupedSubConns[group]
	} else {
		// 没有指定分组，使用所有连接
		for _, conns := range p.groupedSubConns {
			candidates = append(candidates, conns...)
		}
	}

	if len(candidates) == 0 {
		return balancer.PickResult{}, balancer.ErrNoSubConnAvailable
	}

	// 第三步：在候选连接中做轮询
	idx := p.index % uint64(len(candidates))
	p.index++

	return balancer.PickResult{
		SubConn: candidates[idx],
	}, nil
}

// groupPickerBuilder 分组路由 Builder
type groupPickerBuilder struct{}

// Build 实现 PickerBuilder 接口
// 从 SubConnInfo 中提取分组信息
func (b *groupPickerBuilder) Build(readySCs map[balancer.SubConn]balancer.SubConnInfo) balancer.Picker {
	// 按分组组织连接
	groupedConns := make(map[string][]balancer.SubConn)
	for sc, info := range readySCs {
		// 从 Address 的 Attributes 中获取分组信息
		// 分组信息是在服务注册时写入的
		group := getGroupFromAddress(info.Address)
		groupedConns[group] = append(groupedConns[group], sc)
	}

	return &groupPicker{
		groupedSubConns: groupedConns,
	}
}

// getGroupFromAddress 从 resolver.Address 中读取分组信息
// 分组信息是在服务注册阶段写入的
func getGroupFromAddress(addr resolver.Address) string {
	// 实际实现中从 addr.Attributes 获取
	// 这里简化处理
	return ""
}

// ============================================================
// 服务端注册时设置分组
// ============================================================

/*
服务端在启动时通过环境变量或配置文件指定分组：

```go
// 通过环境变量设置分组
group := os.Getenv("SERVICE_GROUP") // "A" 或 "B"

// 注册到 etcd 时，将分组信息写入 Metadata
instance := &registry.ServiceInstance{
	ServiceName: "user-service",
	Address:     "172.10.0.1",
	Port:        9091,
	Metadata: map[string]string{
		"group": group, // 分组信息
	},
}

// 设置分组为 A，此请求只会发送到 A 组的实例
ctx := WithGroup(context.Background(), "A")
resp, err := client.GetUser(ctx, &GetUserRequest{Id: 123})

// 设置分组为 B，此请求只会发送到 B 组的实例
ctx = WithGroup(context.Background(), "B")
resp, err = client.GetUser(ctx, &GetUserRequest{Id: 456})

// failfast 实际上就是默认行为，不设置 retryPolicy 即可
// 也可以在调用时通过 grpc.FailFast(true) 明确指定（这是默认行为）

// const FailfastServiceConfig = `{}`
// 不设置 retryPolicy 就是 failfast

package cluster

import (
	"context"
	"reflect"
	"sync"

	"google.golang.org/grpc"
)

// ============================================================
// gRPC 广播拦截器实现
// ============================================================

// broadcastKey 用于在 context 中标记广播请求
type broadcastKey struct{}

// WithBroadcast 在 context 中设置广播标记
// 只有设置了此标记的请求才会执行广播逻辑
func WithBroadcast(ctx context.Context) context.Context {
	return context.WithValue(ctx, broadcastKey{}, true)
}

// isBroadcast 检查是否为广播请求
func isBroadcast(ctx context.Context) bool {
	v, ok := ctx.Value(broadcastKey{}).(bool)
	return ok && v
}

// BroadcastInterceptor 广播拦截器
// 参数 registry 是注册中心实例，用于获取所有服务端节点
// 参数 newClient 是创建到指定实例的 gRPC 客户端的函数
func BroadcastInterceptor(
	registry Registry,        // 注册中心，用于获取所有实例
	newClient func(string) (*grpc.ClientConn, error), // 根据地址创建连接
) grpc.UnaryClientInterceptor {
	return func(
		ctx context.Context,       // 请求上下文
		method string,             // 调用的方法名，如 "/user.UserService/GetUser"
		req, reply interface{},    // 请求和响应
		cc *grpc.ClientConn,       // 客户端连接（广播时不使用）
		invoker grpc.UnaryInvoker, // 原始调用函数
		opts ...grpc.CallOption,   // 调用选项
	) error {
		// 第一步：检查是否为广播请求
		if !isBroadcast(ctx) {
			// 不是广播请求，正常调用
			return invoker(ctx, method, req, reply, cc, opts...)
		}

		// 第二步：从注册中心获取所有服务实例
		// target 是当前连接的服务名，需要从 cc 中解析
		serviceName := getServiceName(cc)
		instances, err := registry.ListServices(ctx, serviceName)
		if err != nil {
			return err
		}

		// 第三步：并发调用所有实例
		var wg sync.WaitGroup
		errs := make([]error, len(instances))

		for i, inst := range instances {
			wg.Add(1)
			go func(idx int, addr string) {
				defer wg.Done()

				// 为每个实例创建独立的连接
				// 注意：这里的连接难以复用，所以广播性能不好
				conn, err := newClient(addr)
				if err != nil {
					errs[idx] = err
					return
				}
				defer conn.Close()

				// 为每个实例创建独立的响应对象
				// 使用反射创建新对象，避免覆盖问题
				individualReply := reflect.New(reflect.ValueOf(reply).Elem().Type()).Interface()

				// 发起调用
				err = invoker(ctx, method, req, individualReply, conn, opts...)
				errs[idx] = err
			}(i, inst.Address+":"+intToStr(inst.Port))
		}

		wg.Wait()

		// 第四步：处理响应
		// 广播模式下，通常只返回第一个成功的响应
		// 也可以选择返回所有响应或丢弃所有响应
		for _, e := range errs {
			if e == nil {
				return nil // 至少一个成功就返回成功
			}
		}

		// 所有实例都失败了，返回最后一个错误
		if len(errs) > 0 {
			return errs[len(errs)-1]
		}
		return nil
	}
}

// ============================================================
// 广播响应处理策略
// ============================================================

// BroadcastResponseStrategy 广播响应处理策略
type BroadcastResponseStrategy int

const (
	// StrategyDiscard 直接丢弃响应（适用于通知类广播，如缓存刷新）
	StrategyDiscard BroadcastResponseStrategy = iota
	// StrategyFirst 返回最先收到的响应（适用于寻找最快节点）
	StrategyFirst
	// StrategyAll 返回所有响应（适用于需要汇总结果的场景）
	StrategyAll
)

// ============================================================
// 使用 channel 返回所有响应的实现
// ============================================================

// BroadcastWithChannel 使用 channel 返回所有广播响应
// 调用方可以从 channel 中读取每个实例的响应
func BroadcastWithChannel(
	ctx context.Context,
	registry Registry,
	method string,
	req interface{},
	newClient func(string) (*grpc.ClientConn, error),
	invoker grpc.UnaryInvoker,
) (<-chan interface{}, error) {
	// 获取所有实例
	serviceName := ctx.Value("serviceName").(string)
	instances, err := registry.ListServices(ctx, serviceName)
	if err != nil {
		return nil, err
	}

	// 创建响应 channel，缓冲大小等于实例数量
	ch := make(chan interface{}, len(instances))

	go func() {
		defer close(ch) // 重要：关闭 channel，让接收方知道没有更多数据了

		var wg sync.WaitGroup
		for _, inst := range instances {
			wg.Add(1)
			go func(addr string) {
				defer wg.Done()

				conn, err := newClient(addr)
				if err != nil {
					return
				}
				defer conn.Close()

				// 使用反射创建新的响应对象
				reply := reflect.New(reflect.TypeOf(req).Elem()).Interface()
				_ = invoker(ctx, method, req, reply, conn)

				// 将响应写入 channel
				// 注意：这里不会阻塞，因为 channel 有足够的缓冲
				ch <- reply
			}(inst.Address)
		}

		wg.Wait()
	}()

	return ch, nil
}

// 辅助函数
func getServiceName(cc *grpc.ClientConn) string {
	// 从 ClientConn 中解析服务名
	// 实际实现取决于具体的服务发现方案
	return ""
}

func intToStr(n int) string {
	if n == 0 {
		return "0"
	}
	var buf []byte
	if n < 0 {
		buf = append(buf, '-')
		n = -n
	}
	var digits []byte
	for n > 0 {
		digits = append([]byte{byte('0' + n%10)}, digits...)
		n /= 10
	}
	buf = append(buf, digits...)
	return string(buf)
}