8.7 服务治理与容错机制¶

package main

import (
    "encoding/json"
    "fmt"
    "log"
    "net/http"
    "sync"
    "time"
)

// Service 表示一个服务实例
type Service struct {
    Name    string `json:"name"`
    Address string `json:"address"`
    Status  string `json:"status"` // healthy/unhealthy
    LastTTL time.Time `json:"last_ttl"`
}

// Registry 服务注册表
type Registry struct {
    services map[string][]Service // key: 服务名, value: 服务实例列表
    mu       sync.RWMutex
}

func NewRegistry() *Registry {
    return &Registry{
        services: make(map[string][]Service),
    }
}

// Register 注册服务
func (r *Registry) Register(service Service) {
    r.mu.Lock()
    defer r.mu.Unlock()

    service.Status = "healthy"
    service.LastTTL = time.Now().Add(30 * time.Second) // TTL 30秒

    r.services[service.Name] = append(r.services[service.Name], service)
}

// Unregister 注销服务
func (r *Registry) Unregister(name, address string) {
    r.mu.Lock()
    defer r.mu.Unlock()

    if instances, ok := r.services[name]; ok {
        newInstances := []Service{}
        for _, instance := range instances {
            if instance.Address != address {
                newInstances = append(newInstances, instance)
            }
        }
        r.services[name] = newInstances
    }
}

// GetServices 获取服务实例列表
func (r *Registry) GetServices(name string) []Service {
    r.mu.RLock()
    defer r.mu.RUnlock()

    return r.services[name]
}

// Renew 续约服务
func (r *Registry) Renew(name, address string) bool {
    r.mu.Lock()
    defer r.mu.Unlock()

    if instances, ok := r.services[name]; ok {
        for i, instance := range instances {
            if instance.Address == address {
                r.services[name][i].LastTTL = time.Now().Add(30 * time.Second)
                r.services[name][i].Status = "healthy"
                return true
            }
        }
    }
    return false
}

// 清理过期服务
func (r *Registry) cleanupExpiredServices() {
    ticker := time.NewTicker(10 * time.Second)
    defer ticker.Stop()

    for {
        select {
        case <-ticker.C:
            r.mu.Lock()
            for name, instances := range r.services {
                healthyInstances := []Service{}
                for _, instance := range instances {
                    if time.Now().Before(instance.LastTTL) {
                        healthyInstances = append(healthyInstances, instance)
                    }
                }
                r.services[name] = healthyInstances
            }
            r.mu.Unlock()
        }
    }
}

func main() {
    registry := NewRegistry()

    // 启动清理过期服务的协程
    go registry.cleanupExpiredServices()

    // 注册服务
    http.HandleFunc("/register", func(w http.ResponseWriter, r *http.Request) {
        var service Service
        if err := json.NewDecoder(r.Body).Decode(&service); err != nil {
            http.Error(w, err.Error(), http.StatusBadRequest)
            return
        }

        registry.Register(service)
        w.WriteHeader(http.StatusOK)
        fmt.Fprintf(w, "Service %s registered", service.Name)
    })

    // 注销服务
    http.HandleFunc("/unregister", func(w http.ResponseWriter, r *http.Request) {
        name := r.URL.Query().Get("name")
        address := r.URL.Query().Get("address")

        registry.Unregister(name, address)
        w.WriteHeader(http.StatusOK)
        fmt.Fprintf(w, "Service %s unregistered", name)
    })

    // 获取服务
    http.HandleFunc("/services", func(w http.ResponseWriter, r *http.Request) {
        name := r.URL.Query().Get("name")
        services := registry.GetServices(name)

        w.Header().Set("Content-Type", "application/json")
        json.NewEncoder(w).Encode(services)
    })

    // 服务续约
    http.HandleFunc("/renew", func(w http.ResponseWriter, r *http.Request) {
        name := r.URL.Query().Get("name")
        address := r.URL.Query().Get("address")

        success := registry.Renew(name, address)
        if success {
            w.WriteHeader(http.StatusOK)
            fmt.Fprintf(w, "Service %s renewed", name)
        } else {
            http.Error(w, "Service not found", http.StatusNotFound)
        }
    })

    log.Println("Registry server starting on :8500")
    log.Fatal(http.ListenAndServe(":8500", nil))
}

package main

import (
    "errors"
    "fmt"
    "sync"
    "time"
)

// State 熔断器状态
type State int

const (
    StateClosed State = iota
    StateOpen
    StateHalfOpen
)

// CircuitBreaker 熔断器结构
type CircuitBreaker struct {
    state State

    failureThreshold  int           // 失败阈值
    successThreshold  int           // 成功阈值
    timeout           time.Duration // 熔断超时时间
    resetTimeout      time.Duration // 重置超时时间
    lastFailureTime   time.Time     // 最后一次失败时间
    failureCount      int           // 失败计数
    consecutiveSuccesses int        // 连续成功计数

    mutex sync.Mutex
}

// NewCircuitBreaker 创建熔断器
func NewCircuitBreaker(failureThreshold, successThreshold int, timeout, resetTimeout time.Duration) *CircuitBreaker {
    return &CircuitBreaker{
        state:            StateClosed,
        failureThreshold: failureThreshold,
        successThreshold: successThreshold,
        timeout:          timeout,
        resetTimeout:     resetTimeout,
    }
}

// State 获取当前状态
func (cb *CircuitBreaker) State() State {
    cb.mutex.Lock()
    defer cb.mutex.Unlock()
    return cb.state
}

// Execute 执行函数，应用熔断策略
func (cb *CircuitBreaker) Execute(fn func() error) error {
    cb.mutex.Lock()

    // 检查状态并可能转换状态
    switch cb.state {
    case StateOpen:
        // 如果已过重置时间，进入半打开状态
        if time.Since(cb.lastFailureTime) > cb.resetTimeout {
            cb.state = StateHalfOpen
        } else {
            cb.mutex.Unlock()
            return errors.New("circuit breaker is open")
        }
    case StateClosed:
        // 状态不变
    case StateHalfOpen:
        // 状态不变
    }

    currentState := cb.state
    cb.mutex.Unlock()

    // 执行函数并设置超时
    done := make(chan error, 1)
    go func() {
        done <- fn()
    }()

    select {
    case err := <-done:
        cb.handleResult(err)
        return err
    case <-time.After(cb.timeout):
        cb.handleResult(errors.New("timeout"))
        return errors.New("execution timeout")
    }
}

// 处理执行结果，更新熔断器状态
func (cb *CircuitBreaker) handleResult(err error) {
    cb.mutex.Lock()
    defer cb.mutex.Unlock()

    if err != nil {
        // 处理失败情况
        cb.failureCount++
        cb.lastFailureTime = time.Now()
        cb.consecutiveSuccesses = 0

        switch cb.state {
        case StateClosed:
            // 如果失败次数达到阈值，进入打开状态
            if cb.failureCount >= cb.failureThreshold {
                cb.state = StateOpen
            }
        case StateHalfOpen:
            // 半打开状态下任何失败都回到打开状态
            cb.state = StateOpen
        }
    } else {
        // 处理成功情况
        cb.failureCount = 0
        cb.consecutiveSuccesses++

        switch cb.state {
        case StateHalfOpen:
            // 如果连续成功次数达到阈值，进入关闭状态
            if cb.consecutiveSuccesses >= cb.successThreshold {
                cb.state = StateClosed
                cb.consecutiveSuccesses = 0
            }
        }
    }
}

// 示例使用
func main() {
    // 创建熔断器：失败3次后熔断，5秒后尝试恢复，成功2次后关闭熔断，超时1秒
    cb := NewCircuitBreaker(3, 2, 1*time.Second, 5*time.Second)

    // 模拟一个可能失败的函数
    failureRate := 0.7 // 70%的失败率
    attempts := 20

    for i := 0; i < attempts; i++ {
        err := cb.Execute(func() error {
            // 模拟随机失败
            if rand.Float64() < failureRate {
                return errors.New("operation failed")
            }
            return nil
        })

        state := ""
        switch cb.State() {
        case StateClosed:
            state = "Closed"
        case StateOpen:
            state = "Open"
        case StateHalfOpen:
            state = "Half-Open"
        }

        result := "Success"
        if err != nil {
            result = "Failed: " + err.Error()
        }

        fmt.Printf("Attempt %d: %-10s State: %s\n", i+1, result, state)
        time.Sleep(500 * time.Millisecond)
    }
}

package main

import (
    "errors"
    "fmt"
    "net/http"
    "time"

    "github.com/afex/hystrix-go/hystrix"
)

func main() {
    // 配置熔断器
    hystrix.ConfigureCommand("example_command", hystrix.CommandConfig{
        Timeout:                1000, // 超时时间(毫秒)
        MaxConcurrentRequests:  10,   // 最大并发请求数
        ErrorPercentThreshold:  50,   // 错误百分比阈值
        SleepWindow:            5000, // 熔断后多久尝试半打开(毫秒)
        RequestVolumeThreshold: 5,    // 统计错误率的请求数阈值
    })

    // 注册一个简单的HTTP处理函数来演示熔断器
    http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
        // 使用熔断器执行可能失败的操作
        err := hystrix.Do("example_command", func() error {
            // 模拟一个可能失败的服务调用
            return callExternalService()
        }, func(err error) error {
            // 降级函数：当熔断器打开或执行失败时调用
            fmt.Println("Fallback triggered:", err)
            w.WriteHeader(http.StatusServiceUnavailable)
            fmt.Fprintf(w, "Service temporarily unavailable, please try again later")
            return nil
        })

        if err != nil {
            w.WriteHeader(http.StatusInternalServerError)
            fmt.Fprintf(w, "Error: %v", err)
        } else {
            w.WriteHeader(http.StatusOK)
            fmt.Fprintf(w, "Successfully processed request")
        }
    })

    fmt.Println("Server starting on :8080")
    http.ListenAndServe(":8080", nil)
}

// 模拟调用外部服务，有50%的失败概率
func callExternalService() error {
    // 模拟网络延迟
    time.Sleep(200 * time.Millisecond)

    // 50%的概率返回错误
    if time.Now().UnixNano()%2 == 0 {
        return errors.New("external service error")
    }

    return nil
}

package main

import (
    "errors"
    "fmt"
    "sync"
    "time"
)

// TokenBucket 令牌桶
type TokenBucket struct {
    capacity     int           // 桶容量
    rate         int           // 令牌生成速率(个/秒)
    tokens       int           // 当前令牌数
    lastRefill   time.Time     // 最后一次填充令牌的时间
    mutex        sync.Mutex
}

// NewTokenBucket 创建令牌桶
func NewTokenBucket(capacity, rate int) *TokenBucket {
    return &TokenBucket{
        capacity:   capacity,
        rate:       rate,
        tokens:     capacity, // 初始填满令牌
        lastRefill: time.Now(),
    }
}

// refill 填充令牌
func (tb *TokenBucket) refill() {
    now := time.Now()
    elapsed := now.Sub(tb.lastRefill)

    // 计算这段时间应该生成的令牌数
    newTokens := int(elapsed.Seconds()) * tb.rate

    if newTokens > 0 {
        tb.tokens = tb.tokens + newTokens
        // 不能超过桶容量
        if tb.tokens > tb.capacity {
            tb.tokens = tb.capacity
        }
        tb.lastRefill = now
    }
}

// Take 获取令牌
func (tb *TokenBucket) Take() error {
    tb.mutex.Lock()
    defer tb.mutex.Unlock()

    tb.refill()

    if tb.tokens > 0 {
        tb.tokens--
        return nil
    }

    return errors.New("no tokens available")
}

// TakeN 获取多个令牌
func (tb *TokenBucket) TakeN(n int) error {
    tb.mutex.Lock()
    defer tb.mutex.Unlock()

    tb.refill()

    if tb.tokens >= n {
        tb.tokens -= n
        return nil
    }

    return errors.New("not enough tokens available")
}

// Available 获取当前可用令牌数
func (tb *TokenBucket) Available() int {
    tb.mutex.Lock()
    defer tb.mutex.Unlock()

    tb.refill()
    return tb.tokens
}

// 示例使用
func main() {
    // 创建一个容量为10，速率为5个/秒的令牌桶
    tb := NewTokenBucket(10, 5)

    // 模拟请求
    for i := 0; i < 20; i++ {
        err := tb.Take()
        if err != nil {
            fmt.Printf("Request %d: Rejected - %v\n", i+1, err)
        } else {
            fmt.Printf("Request %d: Accepted - Available tokens: %d\n", i+1, tb.Available())
        }

        // 前10个请求快速发送，后面的请求间隔300ms
        if i < 10 {
            time.Sleep(100 * time.Millisecond)
        } else {
            time.Sleep(300 * time.Millisecond)
        }
    }
}

package main

import (
    "errors"
    "fmt"
    "sync"
    "time"
)

// LeakyBucket 漏桶
type LeakyBucket struct {
    capacity     int           // 桶容量
    rate         int           // 漏出速率(个/秒)
    requests     int           // 当前请求数
    lastLeak     time.Time     // 最后一次漏出时间
    mutex        sync.Mutex
}

// NewLeakyBucket 创建漏桶
func NewLeakyBucket(capacity, rate int) *LeakyBucket {
    return &LeakyBucket{
        capacity: capacity,
        rate:     rate,
        requests: 0,
        lastLeak: time.Now(),
    }
}

// leak 漏出请求
func (lb *LeakyBucket) leak() {
    now := time.Now()
    elapsed := now.Sub(lb.lastLeak)

    // 计算这段时间可以漏出的请求数
    leaked := int(elapsed.Seconds()) * lb.rate

    if leaked > 0 {
        if lb.requests > leaked {
            lb.requests -= leaked
        } else {
            lb.requests = 0
        }
        lb.lastLeak = now
    }
}

// Add 添加请求
func (lb *LeakyBucket) Add() error {
    lb.mutex.Lock()
    defer lb.mutex.Unlock()

    lb.leak()

    if lb.requests < lb.capacity {
        lb.requests++
        return nil
    }

    return errors.New("bucket is full")
}

// Pending 获取当前等待的请求数
func (lb *LeakyBucket) Pending() int {
    lb.mutex.Lock()
    defer lb.mutex.Unlock()

    lb.leak()
    return lb.requests
}

// 示例使用
func main() {
    // 创建一个容量为5，速率为2个/秒的漏桶
    lb := NewLeakyBucket(5, 2)

    // 模拟突发请求
    for i := 0; i < 10; i++ {
        err := lb.Add()
        if err != nil {
            fmt.Printf("Request %d: Rejected - %v\n", i+1, err)
        } else {
            fmt.Printf("Request %d: Accepted - Pending requests: %d\n", i+1, lb.Pending())
        }

        // 前5个请求快速发送，后面的请求间隔500ms
        if i < 5 {
            time.Sleep(100 * time.Millisecond)
        } else {
            time.Sleep(500 * time.Millisecond)
        }
    }
}

package main

import (
    "errors"
    "fmt"
    "sync"
    "time"
)

// SlidingWindow 滑动窗口
type SlidingWindow struct {
    windowSize    time.Duration  // 窗口大小
    intervalCount int            // 区间数量
    maxRequests   int            // 最大请求数
    intervals     []int          // 每个区间的请求数
    currentIndex  int            // 当前区间索引
    lastUpdate    time.Time      // 最后更新时间
    mutex         sync.Mutex
}

// NewSlidingWindow 创建滑动窗口
func NewSlidingWindow(windowSize time.Duration, intervalCount, maxRequests int) *SlidingWindow {
    return &SlidingWindow{
        windowSize:    windowSize,
        intervalCount: intervalCount,
        maxRequests:   maxRequests,
        intervals:     make([]int, intervalCount),
        currentIndex:  0,
        lastUpdate:    time.Now(),
    }
}

// 计算当前区间索引并清理过期区间
func (sw *SlidingWindow) updateCurrentInterval() {
    now := time.Now()
    elapsed := now.Sub(sw.lastUpdate)

    // 每个区间的时长
    intervalDuration := sw.windowSize / time.Duration(sw.intervalCount)

    // 计算需要移动的区间数
    intervalsToMove := int(elapsed / intervalDuration)

    if intervalsToMove > 0 {
        // 移动窗口，清理过期区间
        for i := 0; i < intervalsToMove && i < sw.intervalCount; i++ {
            sw.currentIndex = (sw.currentIndex + 1) % sw.intervalCount
            sw.intervals[sw.currentIndex] = 0
        }
        sw.lastUpdate = now
    }
}

// Allow 检查是否允许请求
func (sw *SlidingWindow) Allow() bool {
    sw.mutex.Lock()
    defer sw.mutex.Unlock()

    sw.updateCurrentInterval()

    // 计算总请求数
    total := 0
    for _, count := range sw.intervals {
        total += count
    }

    if total < sw.maxRequests {
        sw.intervals[sw.currentIndex]++
        return true
    }

    return false
}

// Count 获取当前窗口内的请求数
func (sw *SlidingWindow) Count() int {
    sw.mutex.Lock()
    defer sw.mutex.Unlock()

    sw.updateCurrentInterval()

    total := 0
    for _, count := range sw.intervals {
        total += count
    }
    return total
}

// 示例使用
func main() {
    // 创建一个10秒窗口，分为10个区间，最大15个请求
    sw := NewSlidingWindow(10*time.Second, 10, 15)

    // 模拟请求
    for i := 0; i < 30; i++ {
        allowed := sw.Allow()
        if allowed {
            fmt.Printf("Request %d: Allowed - Current count: %d\n", i+1, sw.Count())
        } else {
            fmt.Printf("Request %d: Rejected - Current count: %d\n", i+1, sw.Count())
        }

        // 前10个请求快速发送，后面的请求间隔1秒
        if i < 10 {
            time.Sleep(200 * time.Millisecond)
        } else {
            time.Sleep(1 * time.Second)
        }
    }
}

package main

import (
    "context"
    "fmt"
    "time"

    "github.com/go-redis/redis/v8"
)

// RedisRateLimiter 基于Redis的分布式限流器
type RedisRateLimiter struct {
    client     *redis.Client
    key        string        // Redis键名
    rate       int           // 速率(个/秒)
    burst      int           // 突发容量
    windowSize time.Duration // 窗口大小
}

// NewRedisRateLimiter 创建分布式限流器
func NewRedisRateLimiter(client *redis.Client, key string, rate int, burst int, windowSize time.Duration) *RedisRateLimiter {
    return &RedisRateLimiter{
        client:     client,
        key:        key,
        rate:       rate,
        burst:      burst,
        windowSize: windowSize,
    }
}

// Allow 检查是否允许请求
func (r *RedisRateLimiter) Allow(ctx context.Context) (bool, error) {
    // 使用Redis的INCR命令实现计数器
    count, err := r.client.Incr(ctx, r.key).Result()
    if err != nil {
        return false, err
    }

    // 第一次设置过期时间
    if count == 1 {
        _, err = r.client.Expire(ctx, r.key, r.windowSize).Result()
        if err != nil {
            return false, err
        }
    }

    // 检查是否超过限制
    return count <= int64(r.rate+r.burst), nil
}

// TokenBucketAllow 使用令牌桶算法检查是否允许请求
func (r *RedisRateLimiter) TokenBucketAllow(ctx context.Context) (bool, error) {
    // 使用Redis的令牌桶脚本
    script := `
    local tokens_key = KEYS[1]
    local timestamp_key = KEYS[2]

    local rate = tonumber(ARGV[1])
    local burst = tonumber(ARGV[2])
    local now = tonumber(ARGV[3])
    local window_size = tonumber(ARGV[4])

    -- 计算填充速率：每个窗口填充的令牌数
    local fill_rate = rate * window_size / 1000

    -- 获取当前令牌数和最后更新时间
    local current_tokens = tonumber(redis.call('get', tokens_key) or burst)
    local last_refreshed = tonumber(redis.call('get', timestamp_key) or now)

    -- 计算时间差
    local delta = now - last_refreshed

    -- 计算新的令牌数
    current_tokens = math.min(burst, current_tokens + fill_rate * delta / 1000)
    local new_tokens = current_tokens - 1

    -- 更新令牌数和时间戳
    redis.call('setex', tokens_key, window_size / 1000, new_tokens)
    redis.call('setex', timestamp_key, window_size / 1000, now)

    -- 返回是否允许请求
    return new_tokens >= 0
    `

    now := time.Now().UnixMilli()
    keys := []string{r.key + ":tokens", r.key + ":timestamp"}
    args := []interface{}{r.rate, r.burst, now, r.windowSize.Milliseconds()}

    result, err := r.client.Eval(ctx, script, keys, args...).Bool()
    if err != nil {
        return false, err
    }

    return result, nil
}

// 示例使用
func main() {
    // 连接Redis
    client := redis.NewClient(&redis.Options{
        Addr: "localhost:6379",
    })

    // 创建限流器：10个/秒，突发5个，窗口1000ms
    limiter := NewRedisRateLimiter(client, "api:rate_limit", 10, 5, time.Second)

    ctx := context.Background()

    // 模拟请求
    for i := 0; i < 20; i++ {
        allowed, err := limiter.TokenBucketAllow(ctx)
        if err != nil {
            fmt.Printf("Request %d: Error - %v\n", i+1, err)
        } else if allowed {
            fmt.Printf("Request %d: Allowed\n", i+1)
        } else {
            fmt.Printf("Request %d: Rejected\n", i+1)
        }

        // 控制请求频率
        time.Sleep(100 * time.Millisecond)
    }
}

package main

import (
    "errors"
    "fmt"
    "math/rand"
    "time"
)

// ExponentialBackoff 指数退避重试
func ExponentialBackoff(attempts int, baseDelay time.Duration, maxDelay time.Duration, jitter float64) time.Duration {
    if attempts <= 0 {
        return 0
    }

    // 计算指数延迟: baseDelay * 2^(attempts-1)
    delay := baseDelay * (1 << (attempts - 1))

    // 添加随机抖动，避免重试风暴
    if jitter > 0 {
        rand.Seed(time.Now().UnixNano())
        jitterAmount := time.Duration(rand.Float64() * jitter * float64(delay))
        delay += jitterAmount
    }

    // 不超过最大延迟
    if delay > maxDelay {
        delay = maxDelay
    }

    return delay
}

// RetryWithBackoff 使用指数退避策略重试函数
func RetryWithBackoff(fn func() error, maxAttempts int, baseDelay time.Duration, maxDelay time.Duration, jitter float64) error {
    var lastErr error

    for attempt := 1; attempt <= maxAttempts; attempt++ {
        err := fn()
        if err == nil {
            return nil // 成功
        }

        lastErr = err

        // 如果是最后一次尝试，不再重试
        if attempt == maxAttempts {
            break
        }

        // 计算延迟
        delay := ExponentialBackoff(attempt, baseDelay, maxDelay, jitter)
        fmt.Printf("Attempt %d failed: %v. Retrying in %v...\n", attempt, err, delay)

        time.Sleep(delay)
    }

    return fmt.Errorf("after %d attempts, last error: %w", maxAttempts, lastErr)
}

// 示例使用
func main() {
    // 模拟一个可能失败的操作，第3次尝试成功
    attempts := 0
    fn := func() error {
        attempts++
        fmt.Printf("Performing operation (attempt %d)...\n", attempts)

        // 前2次失败，第3次成功
        if attempts < 3 {
            return errors.New("temporary error")
        }

        return nil
    }

    // 最多重试5次，初始延迟1秒，最大延迟10秒，抖动0.2
    err := RetryWithBackoff(fn, 5, time.Second, 10*time.Second, 0.2)
    if err != nil {
        fmt.Printf("Final error: %v\n", err)
    } else {
        fmt.Println("Operation succeeded!")
    }
}

package main

import (
    "errors"
    "fmt"
    "net/http"
    "time"
)

// 定义可重试的错误类型
var (
    ErrTemporary = errors.New("temporary error")
    ErrTimeout   = errors.New("timeout error")
)

// IsRetryable 判断错误是否可以重试
func IsRetryable(err error) bool {
    if err == nil {
        return false
    }

    // 判断特定错误类型
    switch err {
    case ErrTemporary, ErrTimeout:
        return true
    }

    // 检查HTTP错误状态码
    var httpErr interface{ StatusCode() int }
    if errors.As(err, &httpErr) {
        statusCode := httpErr.StatusCode()
        // 5xx服务器错误通常可以重试
        // 429请求过多也可以重试
        return statusCode >= 500 || statusCode == http.StatusTooManyRequests
    }

    return false
}

// RetryWithCondition 带条件判断的重试
func RetryWithCondition(fn func() error, maxAttempts int, delay time.Duration) error {
    var lastErr error

    for attempt := 1; attempt <= maxAttempts; attempt++ {
        err := fn()
        if err == nil {
            return nil // 成功
        }

        lastErr = err

        // 检查是否可以重试
        if !IsRetryable(err) {
            return fmt.Errorf("non-retryable error: %w", err)
        }

        // 如果是最后一次尝试，不再重试
        if attempt == maxAttempts {
            break
        }

        fmt.Printf("Attempt %d failed: %v. Retrying in %v...\n", attempt, err, delay)
        time.Sleep(delay)
    }

    return fmt.Errorf("after %d attempts, last error: %w", maxAttempts, lastErr)
}

// 示例使用
func main() {
    // 模拟一个可能返回不同错误的操作
    attempts := 0
    fn := func() error {
        attempts++
        fmt.Printf("Performing operation (attempt %d)...\n", attempts)

        // 第一次返回临时错误，第二次返回超时错误，第三次成功
        switch attempts {
        case 1:
            return ErrTemporary
        case 2:
            return ErrTimeout
        default:
            return nil
        }
    }

    // 最多重试3次，每次延迟1秒
    err := RetryWithCondition(fn, 3, time.Second)
    if err != nil {
        fmt.Printf("Final error: %v\n", err)
    } else {
        fmt.Println("Operation succeeded!")
    }
}

package main

import (
    "errors"
    "fmt"
    "sync"
)

// OrderService 订单服务
type OrderService struct {
    orders     map[string]bool // 记录已处理的订单ID
    mutex      sync.Mutex
}

func NewOrderService() *OrderService {
    return &OrderService{
        orders: make(map[string]bool),
    }
}

// CreateOrder 创建订单（幂等操作）
func (s *OrderService) CreateOrder(orderID string, amount float64) error {
    s.mutex.Lock()
    defer s.mutex.Unlock()

    // 检查订单是否已处理
    if s.orders[orderID] {
        fmt.Printf("Order %s already processed\n", orderID)
        return nil // 已处理，直接返回成功
    }

    // 模拟创建订单的业务逻辑
    fmt.Printf("Creating order %s with amount %.2f\n", orderID, amount)

    // 模拟可能的错误
    if amount < 0 {
        return errors.New("invalid amount")
    }

    // 标记订单为已处理
    s.orders[orderID] = true
    return nil
}

// 示例使用
func main() {
    service := NewOrderService()
    orderID := "ORDER_12345"

    // 模拟重试场景
    for i := 0; i < 3; i++ {
        fmt.Printf("Attempt %d to create order...\n", i+1)
        err := service.CreateOrder(orderID, 99.99)
        if err != nil {
            fmt.Printf("Error: %v\n", err)
            break
        }
    }
}