package main
import (
	"fmt"
	"sync"
	"time"
)
// TokenBucket 令牌桶限流
type TokenBucket struct {
	capacity       int           // 桶容量
	tokens         int           // 當(dāng)前令牌數(shù)
	rate           int           // 令牌生成速率（個(gè)/秒）
	lastRefillTime time.Time     // 上次填充時(shí)間
	mutex          sync.Mutex    // 互斥鎖
}
// NewTokenBucket 創(chuàng)建新的令牌桶
func NewTokenBucket(capacity, rate int) *TokenBucket {
	return &TokenBucket{
		capacity:       capacity,
		tokens:         capacity,
		rate:           rate,
		lastRefillTime: time.Now(),
	}
}
// refill 填充令牌
func (tb *TokenBucket) refill() {
	tb.mutex.Lock()
	defer tb.mutex.Unlock()
	now := time.Now()
	timeElapsed := now.Sub(tb.lastRefillTime).Seconds()
	tokensToAdd := int(timeElapsed * float64(tb.rate))
	if tokensToAdd > 0 {
		tb.tokens = min(tb.capacity, tb.tokens+tokensToAdd)
		tb.lastRefillTime = now
	}
}
// Allow 檢查是否允許請(qǐng)求
func (tb *TokenBucket) Allow() bool {
	tb.refill()
	tb.mutex.Lock()
	defer tb.mutex.Unlock()
	if tb.tokens > 0 {
		tb.tokens--
		return true
	}
	return false
}
func min(a, b int) int {
	if a < b {
		return a
	}
	return b
}
func main() {
	// 創(chuàng)建令牌桶，容量10，速率2個(gè)/秒
	tb := NewTokenBucket(10, 2)
	// 測(cè)試限流
	for i := 0; i < 20; i++ {
		if tb.Allow() {
			fmt.Printf("Request %d: allowed\n", i+1)
		} else {
			fmt.Printf("Request %d: denied\n", i+1)
		}
		time.Sleep(100 * time.Millisecond)
	}
	// 等待令牌生成
	time.Sleep(2 * time.Second)
	// 再次測(cè)試
	fmt.Println("\nAfter waiting 2 seconds:")
	for i := 0; i < 10; i++ {
		if tb.Allow() {
			fmt.Printf("Request %d: allowed\n", i+1)
		} else {
			fmt.Printf("Request %d: denied\n", i+1)
		}
		time.Sleep(100 * time.Millisecond)
	}
}

package main
import (
	"fmt"
	"sync"
	"time"
)
// LeakyBucket 漏桶限流
type LeakyBucket struct {
	capacity       int           // 桶容量
	current        int           // 當(dāng)前水量
	rate           int           // 漏水速率（個(gè)/秒）
	lastLeakTime   time.Time     // 上次漏水時(shí)間
	mutex          sync.Mutex    // 互斥鎖
}
// NewLeakyBucket 創(chuàng)建新的漏桶
func NewLeakyBucket(capacity, rate int) *LeakyBucket {
	return &LeakyBucket{
		capacity:     capacity,
		current:      0,
		rate:         rate,
		lastLeakTime: time.Now(),
	}
}
// leak 漏水
func (lb *LeakyBucket) leak() {
	lb.mutex.Lock()
	defer lb.mutex.Unlock()
	now := time.Now()
	timeElapsed := now.Sub(lb.lastLeakTime).Seconds()
	waterToLeak := int(timeElapsed * float64(lb.rate))
	if waterToLeak > 0 {
		lb.current = max(0, lb.current-waterToLeak)
		lb.lastLeakTime = now
	}
}
// Allow 檢查是否允許請(qǐng)求
func (lb *LeakyBucket) Allow() bool {
	lb.leak()
	lb.mutex.Lock()
	defer lb.mutex.Unlock()
	if lb.current < lb.capacity {
		lb.current++
		return true
	}
	return false
}
func max(a, b int) int {
	if a > b {
		return a
	}
	return b
}
func main() {
	// 創(chuàng)建漏桶，容量5，速率1個(gè)/秒
	lb := NewLeakyBucket(5, 1)
	// 測(cè)試限流
	for i := 0; i < 15; i++ {
		if lb.Allow() {
			fmt.Printf("Request %d: allowed\n", i+1)
		} else {
			fmt.Printf("Request %d: denied\n", i+1)
		}
		time.Sleep(200 * time.Millisecond)
	}
	// 等待漏水
	time.Sleep(3 * time.Second)
	// 再次測(cè)試
	fmt.Println("\nAfter waiting 3 seconds:")
	for i := 0; i < 10; i++ {
		if lb.Allow() {
			fmt.Printf("Request %d: allowed\n", i+1)
		} else {
			fmt.Printf("Request %d: denied\n", i+1)
		}
		time.Sleep(200 * time.Millisecond)
	}
}

package main
import (
	"fmt"
	"sync"
	"time"
)
// SlidingWindow 滑動(dòng)窗口限流
type SlidingWindow struct {
	windowSize     time.Duration // 窗口大小
	maxRequests    int           // 最大請(qǐng)求數(shù)
	requests       []time.Time   // 請(qǐng)求時(shí)間戳
	mutex          sync.Mutex    // 互斥鎖
}
// NewSlidingWindow 創(chuàng)建新的滑動(dòng)窗口
func NewSlidingWindow(windowSize time.Duration, maxRequests int) *SlidingWindow {
	return &SlidingWindow{
		windowSize:  windowSize,
		maxRequests: maxRequests,
		requests:    make([]time.Time, 0),
	}
}
// Allow 檢查是否允許請(qǐng)求
func (sw *SlidingWindow) Allow() bool {
	sw.mutex.Lock()
	defer sw.mutex.Unlock()
	now := time.Now()
	// 清理過(guò)期的請(qǐng)求
	var validRequests []time.Time
	for _, reqTime := range sw.requests {
		if now.Sub(reqTime) < sw.windowSize {
			validRequests = append(validRequests, reqTime)
		}
	}
	sw.requests = validRequests
	// 檢查請(qǐng)求數(shù)是否超過(guò)閾值
	if len(sw.requests) < sw.maxRequests {
		sw.requests = append(sw.requests, now)
		return true
	}
	return false
}
func main() {
	// 創(chuàng)建滑動(dòng)窗口，窗口大小1秒，最大請(qǐng)求數(shù)3
	sw := NewSlidingWindow(time.Second, 3)
	// 測(cè)試限流
	for i := 0; i < 10; i++ {
		if sw.Allow() {
			fmt.Printf("Request %d: allowed\n", i+1)
		} else {
			fmt.Printf("Request %d: denied\n", i+1)
		}
		time.Sleep(200 * time.Millisecond)
	}
	// 等待窗口滑動(dòng)
	time.Sleep(1 * time.Second)
	// 再次測(cè)試
	fmt.Println("\nAfter waiting 1 second:")
	for i := 0; i < 5; i++ {
		if sw.Allow() {
			fmt.Printf("Request %d: allowed\n", i+1)
		} else {
			fmt.Printf("Request %d: denied\n", i+1)
		}
		time.Sleep(200 * time.Millisecond)
	}
}

package main
import (
	"fmt"
	"sync"
	"time"
)
// CircuitState 熔斷器狀態(tài)
type CircuitState int
const (
	StateClosed CircuitState = iota  // 關(guān)閉狀態(tài)
	StateOpen                        // 打開(kāi)狀態(tài)
	StateHalfOpen                    // 半開(kāi)狀態(tài)
)
// CircuitBreaker 熔斷器
type CircuitBreaker struct {
	state              CircuitState // 當(dāng)前狀態(tài)
	failureThreshold   int          // 失敗閾值
	successThreshold   int          // 成功閾值
	resetTimeout       time.Duration // 重置超時(shí)時(shí)間
	lastFailureTime    time.Time    // 上次失敗時(shí)間
	failureCount       int          // 失敗計(jì)數(shù)
	successCount       int          // 成功計(jì)數(shù)
	mutex              sync.Mutex   // 互斥鎖
}
// NewCircuitBreaker 創(chuàng)建新的熔斷器
func NewCircuitBreaker(failureThreshold, successThreshold int, resetTimeout time.Duration) *CircuitBreaker {
	return &CircuitBreaker{
		state:            StateClosed,
		failureThreshold: failureThreshold,
		successThreshold: successThreshold,
		resetTimeout:     resetTimeout,
	}
}
// Allow 檢查是否允許請(qǐng)求
func (cb *CircuitBreaker) Allow() bool {
	cb.mutex.Lock()
	defer cb.mutex.Unlock()
	switch cb.state {
	case StateClosed:
		return true
	case StateOpen:
		// 檢查是否可以進(jìn)入半開(kāi)狀態(tài)
		if time.Since(cb.lastFailureTime) > cb.resetTimeout {
			cb.state = StateHalfOpen
			cb.successCount = 0
			return true
		}
		return false
	case StateHalfOpen:
		return true
	default:
		return true
	}
}
// RecordSuccess 記錄成功
func (cb *CircuitBreaker) RecordSuccess() {
	cb.mutex.Lock()
	defer cb.mutex.Unlock()
	switch cb.state {
	case StateClosed:
		// 重置失敗計(jì)數(shù)
		cb.failureCount = 0
	case StateHalfOpen:
		// 增加成功計(jì)數(shù)
		cb.successCount++
		if cb.successCount >= cb.successThreshold {
			// 成功次數(shù)達(dá)到閾值，關(guān)閉熔斷器
			cb.state = StateClosed
			cb.failureCount = 0
			cb.successCount = 0
		}
	}
}
// RecordFailure 記錄失敗
func (cb *CircuitBreaker) RecordFailure() {
	cb.mutex.Lock()
	defer cb.mutex.Unlock()
	switch cb.state {
	case StateClosed:
		// 增加失敗計(jì)數(shù)
		cb.failureCount++
		if cb.failureCount >= cb.failureThreshold {
			// 失敗次數(shù)達(dá)到閾值，打開(kāi)熔斷器
			cb.state = StateOpen
			cb.lastFailureTime = time.Now()
		}
	case StateHalfOpen:
		// 半開(kāi)狀態(tài)下失敗，重新打開(kāi)熔斷器
		cb.state = StateOpen
		cb.lastFailureTime = time.Now()
		cb.successCount = 0
	}
}
func main() {
	// 創(chuàng)建熔斷器，失敗閾值3，成功閾值2，重置超時(shí)5秒
	cb := NewCircuitBreaker(3, 2, 5*time.Second)
	// 模擬失敗，觸發(fā)熔斷
	fmt.Println("=== Testing failure scenarios ===")
	for i := 0; i < 5; i++ {
		if cb.Allow() {
			fmt.Printf("Attempt %d: allowed, simulating failure\n", i+1)
			cb.RecordFailure()
		} else {
			fmt.Printf("Attempt %d: denied (circuit open)\n", i+1)
		}
		time.Sleep(500 * time.Millisecond)
	}
	// 等待重置超時(shí)
	fmt.Println("\n=== Waiting for reset timeout ===")
	time.Sleep(6 * time.Second)
	// 模擬成功，關(guān)閉熔斷
	fmt.Println("\n=== Testing success scenarios ===")
	for i := 0; i < 5; i++ {
		if cb.Allow() {
			fmt.Printf("Attempt %d: allowed, simulating success\n", i+1)
			cb.RecordSuccess()
		} else {
			fmt.Printf("Attempt %d: denied\n", i+1)
		}
		time.Sleep(500 * time.Millisecond)
	}
	// 再次模擬失敗
	fmt.Println("\n=== Testing failure scenarios again ===")
	for i := 0; i < 4; i++ {
		if cb.Allow() {
			fmt.Printf("Attempt %d: allowed, simulating failure\n", i+1)
			cb.RecordFailure()
		} else {
			fmt.Printf("Attempt %d: denied (circuit open)\n", i+1)
		}
		time.Sleep(500 * time.Millisecond)
	}
}

package main
import (
	"fmt"
	"net/http"
	"time"
	"github.com/afex/hystrix-go/hystrix"
)
func main() {
	// 配置熔斷器
	hystrix.ConfigureCommand("api_request", hystrix.CommandConfig{
		Timeout:                1000,                // 超時(shí)時(shí)間（毫秒）
		MaxConcurrentRequests:  100,                 // 最大并發(fā)請(qǐng)求數(shù)
		ErrorThresholdPercentage: 25,                // 錯(cuò)誤閾值百分比
		SleepWindow:            5000,                // 睡眠窗口（毫秒）
		RequestVolumeThreshold:  5,                  // 請(qǐng)求 volume 閾值
	})
	// 模擬API請(qǐng)求
	for i := 0; i < 20; i++ {
		var response string
		err := hystrix.Do("api_request", func() error {
			// 模擬API調(diào)用
			if i%3 == 0 {
				// 模擬失敗
				return fmt.Errorf("API error")
			}
			// 模擬成功
			response = "API response"
			return nil
		}, func(err error) error {
			// 降級(jí)處理
			response = "Fallback response"
			return nil
		})
		fmt.Printf("Request %d: %s, error: %v\n", i+1, response, err)
		time.Sleep(200 * time.Millisecond)
	}
}

package main
import (
	"fmt"
	"net/http"
	"sync"
	"time"
)
// RateLimiter 速率限制器
type RateLimiter struct {
	tokenBuckets map[string]*TokenBucket // 每個(gè)IP的令牌桶
	mutex        sync.Mutex
}
// NewRateLimiter 創(chuàng)建速率限制器
func NewRateLimiter() *RateLimiter {
	return &RateLimiter{
		tokenBuckets: make(map[string]*TokenBucket),
	}
}
// getTokenBucket 獲取或創(chuàng)建令牌桶
func (rl *RateLimiter) getTokenBucket(ip string) *TokenBucket {
	rl.mutex.Lock()
	defer rl.mutex.Unlock()
	if tb, exists := rl.tokenBuckets[ip]; exists {
		return tb
	}
	// 為每個(gè)IP創(chuàng)建一個(gè)令牌桶，容量10，速率2個(gè)/秒
	tb := NewTokenBucket(10, 2)
	rl.tokenBuckets[ip] = tb
	return tb
}
// RateLimitMiddleware 限流中間件
func RateLimitMiddleware(rl *RateLimiter) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		// 獲取客戶端IP
		ip := r.RemoteAddr
		// 檢查是否允許請(qǐng)求
		if !rl.getTokenBucket(ip).Allow() {
			http.Error(w, "Too Many Requests", http.StatusTooManyRequests)
			return
		}
		// 處理請(qǐng)求
		w.WriteHeader(http.StatusOK)
		w.Write([]byte("Hello, World!"))
	}
}
func main() {
	rl := NewRateLimiter()
	http.HandleFunc("/", RateLimitMiddleware(rl))
	fmt.Println("Server started on :8080")
	http.ListenAndServe(":8080", nil)
}

package main
import (
	"fmt"
	"net/http"
	"time"
	"github.com/afex/hystrix-go/hystrix"
)
// ServiceClient 服務(wù)客戶端
type ServiceClient struct {
	baseURL string
}
// NewServiceClient 創(chuàng)建服務(wù)客戶端
func NewServiceClient(baseURL string) *ServiceClient {
	return &ServiceClient{baseURL: baseURL}
}
// CallService 調(diào)用服務(wù)
func (sc *ServiceClient) CallService(endpoint string) (string, error) {
	var response string
	err := hystrix.Do("service_call", func() error {
		// 模擬服務(wù)調(diào)用
		url := sc.baseURL + endpoint
		resp, err := http.Get(url)
		if err != nil {
			return err
		}
		defer resp.Body.Close()
		if resp.StatusCode != http.StatusOK {
			return fmt.Errorf("service returned status %d", resp.StatusCode)
		}
		// 讀取響應(yīng)
		// ...
		response = "Service response"
		return nil
	}, func(err error) error {
		// 降級(jí)處理
		response = "Fallback response"
		return nil
	})
	return response, err
}
func main() {
	// 配置熔斷器
	hystrix.ConfigureCommand("service_call", hystrix.CommandConfig{
		Timeout:                1000,
		MaxConcurrentRequests:  100,
		ErrorThresholdPercentage: 25,
		SleepWindow:            5000,
		RequestVolumeThreshold:  5,
	})
	client := NewServiceClient("http://localhost:8081")
	// 模擬服務(wù)調(diào)用
	for i := 0; i < 20; i++ {
		response, err := client.CallService("/api")
		fmt.Printf("Call %d: %s, error: %v\n", i+1, response, err)
		time.Sleep(200 * time.Millisecond)
	}
}

func (tb *TokenBucket) Allow() bool {
	tb.refill()
	tb.mutex.Lock()
	defer tb.mutex.Unlock()
	if tb.tokens > 0 {
		tb.tokens--
		return true
	}
	return false
}

func (tb *TokenBucket) Allow() bool {
	tb.refill()
	// 使用原子操作減少鎖的競(jìng)爭(zhēng)
	for {
		current := atomic.LoadInt32(&tb.tokens)
		if current <= 0 {
			return false
		}
		if atomic.CompareAndSwapInt32(&tb.tokens, current, current-1) {
			return true
		}
	}
}

func (cb *CircuitBreaker) Allow() bool {
	cb.mutex.Lock()
	defer cb.mutex.Unlock()
	switch cb.state {
	case StateClosed:
		return true
	case StateOpen:
		if time.Since(cb.lastFailureTime) > cb.resetTimeout {
			cb.state = StateHalfOpen
			cb.successCount = 0
			return true
		}
		return false
	case StateHalfOpen:
		return true
	default:
		return true
	}
}

func (cb *CircuitBreaker) Allow() bool {
	// 快速路徑：如果是關(guān)閉狀態(tài)，直接返回
	if atomic.LoadInt32((*int32)(&cb.state)) == int32(StateClosed) {
		return true
	}
	cb.mutex.Lock()
	defer cb.mutex.Unlock()
	switch cb.state {
	case StateClosed:
		return true
	case StateOpen:
		if time.Since(cb.lastFailureTime) > cb.resetTimeout {
			cb.state = StateHalfOpen
			cb.successCount = 0
			return true
		}
		return false
	case StateHalfOpen:
		return true
	default:
		return true
	}
}

// 本地限流實(shí)現(xiàn)
func (rl *RateLimiter) getTokenBucket(ip string) *TokenBucket {
	rl.mutex.Lock()
	defer rl.mutex.Unlock()
	if tb, exists := rl.tokenBuckets[ip]; exists {
		return tb
	}
	tb := NewTokenBucket(10, 2)
	rl.tokenBuckets[ip] = tb
	return tb
}

// 分布式限流實(shí)現(xiàn)
func (rl *DistributedRateLimiter) Allow(ip string) bool {
	// 使用Redis實(shí)現(xiàn)分布式限流
	key := fmt.Sprintf("rate_limit:%s", ip)
	// 使用Redis的令牌桶算法
	// 1. 獲取當(dāng)前令牌數(shù)
	// 2. 如果令牌數(shù)大于0，減少令牌數(shù)并返回允許
	// 3. 否則返回拒絕
	// 具體實(shí)現(xiàn)使用Lua腳本保證原子性
	// ...
	return true
}