Agent路由入门指南：从零构建高可用服务分发系统

1次阅读

共计 2476 个字符，预计需要花费 7 分钟才能阅读完成。

在微服务架构中，传统负载均衡器（如 Nginx、HAProxy）虽然能提供基本的流量分发功能，但在动态服务发现和故障处理方面存在明显短板：

配置僵化：每次服务实例变动都需要手动更新配置并重启，在容器化环境中尤其致命
故障感知延迟：基于定时健康检查的机制可能导致分钟级的故障切换延迟
协议支持有限：对 gRPC、WebSocket 等新兴协议的支持往往需要复杂配置
缺乏业务感知：无法根据业务属性（如用户地域、请求类型）进行智能路由

特性	Nginx/HAProxy	Agent 路由
动态更新	需手动重载配置	实时生效
健康检查延迟	通常 5 -30 秒	亚秒级
gRPC 支持	需额外模块	原生支持
权重动态调整	不支持	支持
跨协议路由	有限	完整支持

// 基于 Consul 的服务发现与路由
type AgentRouter struct {
    serviceMap sync.Map // 并发安全的服务实例存储
    consul     *api.Client
}

// 初始化 Consul 客户端
func NewRouter(consulAddr string) (*AgentRouter, error) {config := api.DefaultConfig()
    config.Address = consulAddr
    client, err := api.NewClient(config)
    if err != nil {return nil, err}
    return &AgentRouter{consul: client}, nil
}

// 关键点 1：监听服务变化
func (r *AgentRouter) WatchService(serviceName string) {params := &api.QueryOptions{RequireConsistent: true}
    for {services, meta, err := r.consul.Health().Service(serviceName, "", true, params,)
        if err != nil {log.Printf("Watch error: %v", err)
            time.Sleep(2 * time.Second)
            continue
        }

        // 关键点 2：原子更新服务列表
        instances := make([]ServiceInstance, 0, len(services))
        for _, s := range services {
            instances = append(instances, ServiceInstance{
                ID:      s.Service.ID,
                Address: fmt.Sprintf("%s:%d", s.Service.Address, s.Service.Port),
                Weight:  s.Service.Weights.Passing,
            })
        }
        r.serviceMap.Store(serviceName, instances)

        params.WaitIndex = meta.LastIndex // 阻塞查询
    }
}

// 关键点 3：带权重的路由选择
func (r *AgentRouter) Select(serviceName string) (string, error) {v, ok := r.serviceMap.Load(serviceName)
    if !ok {return "", fmt.Errorf("service not found")
    }
    instances := v.([]ServiceInstance)
    if len(instances) == 0 {return "", fmt.Errorf("no available instances")
    }

    // 加权随机算法
    totalWeight := 0
    for _, inst := range instances {totalWeight += inst.Weight}

    rand.Seed(time.Now().UnixNano())
    target := rand.Intn(totalWeight)
    for _, inst := range instances {
        if target < inst.Weight {return inst.Address, nil}
        target -= inst.Weight
    }

    return instances[0].Address, nil // fallback
}

指标	Nginx	Agent 路由
平均延迟	12ms	8ms
P99 延迟	45ms	28ms
CPU 占用	15%	9%
内存占用	320MB	110MB

goroutine 泄漏检测 ：使用go.uber.org/goleak 在单元测试中捕获
连接池管理：为每个下游服务维护独立连接池并设置上限
监控指标 ：暴露runtime.MemStats 指标到 Prometheus

// 连接池示例
type ConnPool struct {
    pool    chan net.Conn
    maxSize int
}

func (p *ConnPool) Get() (net.Conn, error) {
    select {
    case conn := <-p.pool:
        return conn, nil
    default:
        return net.Dial("tcp", "backend:8080")
    }
}

func (p *ConnPool) Put(conn net.Conn) {
    select {
    case p.pool <- conn:
    default: // 超过容量则直接关闭
        conn.Close()}
}