MG 高端 Go 语言：构建高性能微服务治理与可观测性体系

package registry

import (
    "context"
    "fmt"
    "time"
    clientv3 "go.etcd.io/etcd/client/v3"
)

type Registry interface {
    Register(serviceName, instanceId, address string) error
    Deregister(serviceName, instanceId string) error
    Discover(serviceName string) ([]string, error)
    Watch(serviceName string) (<-chan []string, error)
}

type EtcdRegistry struct {
    client *clientv3.Client
    lease  clientv3.Lease
    ttl    int64
}

func NewEtcdRegistry(endpoints []string, ttl int64) (*EtcdRegistry, error) {
    cli, err := clientv3.New(clientv3.Config{
        Endpoints:   endpoints,
        DialTimeout: 5 * time.Second,
    })
    if err != nil {
        return nil, err
    }
    return &EtcdRegistry{client: cli, ttl: ttl}, nil
}

func (e *EtcdRegistry) Register(serviceName, instanceId, address string) error {
    // 创建租约（TTL 自动续约）
    leaseResp, err := e.client.Grant(context.Background(), e.ttl)
    if err != nil {
        return err
    }
    key := fmt.Sprintf("/services/%s/%s", serviceName, instanceId)
    _, err = e.client.Put(context.Background(), key, address, clientv3.WithLease(leaseResp.ID))
    if err != nil {
        return err
    }
    // 保持心跳
    _, err = e.client.KeepAlive(context.Background(), leaseResp.ID)
    return err
}

func (e *EtcdRegistry) Discover(serviceName string) ([]string, error) {
    key := fmt.Sprintf("/services/%s", serviceName)
    resp, err := e.client.Get(context.Background(), key, clientv3.WithPrefix())
    if err != nil {
        return nil, err
    }
    var addrs []string
    for _, kv := range resp.Kvs {
        addrs = append(addrs, string(kv.Value))
    }
    return addrs, nil
}

package client

import (
    "errors"
    "fmt"
    "net/http"
    "sync/atomic"
    "time"
    "github.com/afex/hystrix-go/hystrix"
)

type LoadBalancer interface {
    Next() (string, error)
}

type RoundRobinBalancer struct {
    addrs []string
    index uint64
}

func (lb *RoundRobinBalancer) Next() (string, error) {
    if len(lb.addrs) == 0 {
        return "", errors.New("no available instances")
    }
    idx := atomic.AddUint64(&lb.index, 1) % uint64(len(lb.addrs))
    return lb.addrs[idx], nil
}

// 熔断配置
func initHystrix() {
    hystrix.ConfigureCommand("http_service", hystrix.CommandConfig{
        Timeout:               1000,  // 超时1s
        MaxConcurrentRequests: 100,
        ErrorPercentThreshold: 50,    // 错误率超50%开启熔断
        RequestVolumeThreshold: 5,
        SleepWindow:           5000,  // 半开状态后等待5s
    })
}

// 熔断执行函数
func CallWithCircuitBreaker(url string) (string, error) {
    var result string
    err := hystrix.Do("http_service", func() error {
        resp, err := http.Get(url)
        if err != nil {
            return err
        }
        defer resp.Body.Close()
        // 模拟读取响应
        result = fmt.Sprintf("Status: %s", resp.Status)
        return nil
    }, func(err error) error {
        // 降级逻辑
        return errors.New("service unavailable, fallback response")
    })
    return result, err
}

package tracing

import (
    "context"
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/exporters/jaeger"
    "go.opentelemetry.io/otel/sdk/resource"
    sdktrace "go.opentelemetry.io/otel/sdk/trace"
    semconv "go.opentelemetry.io/otel/semconv/v1.21.0"
)

func InitTracer(serviceName, endpoint string) (*sdktrace.TracerProvider, error) {
    exp, err := jaeger.New(jaeger.WithCollectorEndpoint(jaeger.WithEndpoint(endpoint)))
    if err != nil {
        return nil, err
    }
    tp := sdktrace.NewTracerProvider(
        sdktrace.WithBatcher(exp),
        sdktrace.WithResource(resource.NewWithAttributes(
            semconv.SchemaURL,
            semconv.ServiceName(serviceName),
        )),
        sdktrace.WithSampler(sdktrace.AlwaysSample()),
    )
    otel.SetTracerProvider(tp)
    return tp, nil
}

// 在业务代码中创建 Span
func DoBusiness(ctx context.Context) {
    tracer := otel.Tracer("myapp")
    ctx, span := tracer.Start(ctx, "DoBusiness")
    defer span.End()
    // ... 业务逻辑，可传递上下文至下游
}

package metrics

import (
    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/promhttp"
    "net/http"
)

var (
    RequestCount = prometheus.NewCounterVec(
        prometheus.CounterOpts{
            Name: "http_requests_total",
            Help: "Total HTTP requests",
        },
        []string{"method", "endpoint", "status"},
    )
    RequestDuration = prometheus.NewHistogramVec(
        prometheus.HistogramOpts{
            Name: "http_request_duration_seconds",
            Help: "Histogram of HTTP request durations",
            Buckets: []float64{0.01, 0.05, 0.1, 0.5, 1, 2, 5},
        },
        []string{"method", "endpoint"},
    )
)

func init() {
    prometheus.MustRegister(RequestCount, RequestDuration)
}

// 在中间件中记录指标
func MetricsMiddleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        timer := prometheus.NewTimer(RequestDuration.WithLabelValues(r.Method, r.URL.Path))
        defer timer.ObserveDuration()
        // 包装 ResponseWriter 以捕获状态码
        rw := &statusRecorder{ResponseWriter: w, status: http.StatusOK}
        next.ServeHTTP(rw, r)
        RequestCount.WithLabelValues(r.Method, r.URL.Path, http.StatusText(rw.status)).Inc()
    })
}

type statusRecorder struct {
    http.ResponseWriter
    status int
}
func (s *statusRecorder) WriteHeader(code int) {
    s.status = code
    s.ResponseWriter.WriteHeader(code)
}