red-future/common
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 19 Wiki Activity

增加熔断策略

Browse Source
This commit is contained in:
张斌
2026-01-01 13:54:34 +08:00
parent cb6d0911d3
commit e03a4bfcff
1 changed files with 93 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

157
middleware/circuit_breaker.go
View File

@@ -221,6 +221,43 @@ func (cb *CircuitBreakerInfo) updateStateMetrics(state CircuitBreakerState) {
}
}
// getCircuitBreakerInfoAndConfig 获取熔断器信息和配置
func getCircuitBreakerInfoAndConfig(serviceName string) (*CircuitBreakerInfo, *CircuitBreakerConfig) {
cbInfoVal, ok := circuitBreakers.Load(serviceName)
if !ok {
return nil, nil
}
cbInfo, ok := cbInfoVal.(*CircuitBreakerInfo)
if !ok {
return nil, nil
}
return cbInfo, cbInfo.Config
}
// updateResponseTimeStats 更新响应时间统计
func updateResponseTimeStats(cbInfo *CircuitBreakerInfo, duration time.Duration, config *CircuitBreakerConfig) {
durationNs := duration.Nanoseconds()
cbInfo.Metrics.TotalResponseTime.Add(durationNs)
// 原子更新最小和最大响应时间
atomicUpdateMin(&cbInfo.Metrics.MinResponseTime, durationNs)
atomicUpdateMax(&cbInfo.Metrics.MaxResponseTime, durationNs)
if duration > config.SlowRequestThresholdParsed {
cbInfo.Metrics.SlowRequests.Add(1)
}
}
// formatUnixTime 格式化Unix时间戳
func formatUnixTime(timestamp int64) string {
if timestamp > 0 {
return time.Unix(timestamp, 0).Format("2006-01-02 15:04:05")
}
return ""
}
// InitCircuitBreaker 初始化Sentinel熔断器
func InitCircuitBreaker() error {
ctx := context.Background()
@@ -398,14 +435,24 @@ func atomicUpdateMax(maxValue *atomic.Int64, newValue int64) {
}
}
// getAllowedIPs 获取允许的IP列表（带锁保护）
// getAllowedIPsAndCIDRs 获取允许的IP和CIDR列表（带锁保护）
func getAllowedIPsAndCIDRs() (map[string]bool, []*net.IPNet) {
allowedAdminIPsMutex.RLock()
allowedAdminCIDRsMutex.RLock()
defer allowedAdminIPsMutex.RUnlock()
defer allowedAdminCIDRsMutex.RUnlock()
return allowedAdminIPsMap, allowedAdminCIDRs
}
// getAllowedIPs 获取允许的IP列表（带锁保护，兼容旧代码）
func getAllowedIPs() map[string]bool {
allowedAdminIPsMutex.RLock()
defer allowedAdminIPsMutex.RUnlock()
return allowedAdminIPsMap
}
// getAllowedCIDRs 获取允许的CIDR列表（带锁保护）
// getAllowedCIDRs 获取允许的CIDR列表（带锁保护，兼容旧代码）
func getAllowedCIDRs() []*net.IPNet {
allowedAdminCIDRsMutex.RLock()
defer allowedAdminCIDRsMutex.RUnlock()
@@ -495,41 +542,57 @@ func (cb *CircuitBreakerInfo) updateWindowStats(isSuccess bool, ctx context.Cont
}
}
// validateInRange 验证值是否在指定范围内
func validateInRange(name string, value, min, max int) error {
if value < min || value > max {
return fmt.Errorf("%s必须在%d-%d之间", name, min, max)
}
return nil
}
// validateFloatInRange 验证浮点数值是否在指定范围内
func validateFloatInRange(name string, value, min, max float64) error {
if value < min || value > max {
return fmt.Errorf("%s必须在%.1f-%.1f之间", name, min, max)
}
return nil
}
// validateCircuitBreakerConfig 验证配置
func validateCircuitBreakerConfig(config *CircuitBreakerConfig) error {
if config.MaxFailures <= 0 {
return fmt.Errorf("maxFailures必须大于0")
}
if config.FailureRateThreshold < 0 || config.FailureRateThreshold > 1 {
return fmt.Errorf("failureRateThreshold必须在0.0-1.0之间")
if err := validateFloatInRange("failureRateThreshold", config.FailureRateThreshold, 0.0, 1.0); err != nil {
return err
}
if len(config.SuccessStatusCodes) == 0 {
return fmt.Errorf("successStatusCodes不能为空")
}
if config.RequestTimeout < 0 || config.RequestTimeout > 300000 {
return fmt.Errorf("requestTimeout必须在0-300000毫秒之间")
if err := validateInRange("requestTimeout", config.RequestTimeout, 0, 300000); err != nil {
return err
}
if config.DistributedTTL < 0 || config.DistributedTTL > 3600 {
return fmt.Errorf("distributedTTL必须在0-3600秒之间")
if err := validateInRange("distributedTTL", config.DistributedTTL, 0, 3600); err != nil {
return err
}
if config.StatIntervalMs < 100 || config.StatIntervalMs > 60000 {
return fmt.Errorf("statIntervalMs必须在100-60000毫秒之间")
if err := validateInRange("statIntervalMs", config.StatIntervalMs, 100, 60000); err != nil {
return err
}
if config.MinRequestAmount < 1 || config.MinRequestAmount > 10000 {
return fmt.Errorf("minRequestAmount必须在1-10000之间")
if err := validateInRange("minRequestAmount", config.MinRequestAmount, 1, 10000); err != nil {
return err
}
if config.HalfOpenMaxRequests < 1 || config.HalfOpenMaxRequests > 100 {
return fmt.Errorf("halfOpenMaxRequests必须在1-100之间")
if err := validateInRange("halfOpenMaxRequests", config.HalfOpenMaxRequests, 1, 100); err != nil {
return err
}
if config.HalfOpenSuccessThreshold < 0 || config.HalfOpenSuccessThreshold > 1 {
return fmt.Errorf("halfOpenSuccessThreshold必须在0.0-1.0之间")
if err := validateFloatInRange("halfOpenSuccessThreshold", config.HalfOpenSuccessThreshold, 0.0, 1.0); err != nil {
return err
}
if config.EnableAdaptiveThreshold {
if config.AdaptiveMinThreshold < 0 || config.AdaptiveMinThreshold > 1 {
return fmt.Errorf("adaptiveMinThreshold必须在0.0-1.0之间")
if err := validateFloatInRange("adaptiveMinThreshold", config.AdaptiveMinThreshold, 0.0, 1.0); err != nil {
return err
}
if config.AdaptiveMaxThreshold < 0 || config.AdaptiveMaxThreshold > 1 {
return fmt.Errorf("adaptiveMaxThreshold必须在0.0-1.0之间")
if err := validateFloatInRange("adaptiveMaxThreshold", config.AdaptiveMaxThreshold, 0.0, 1.0); err != nil {
return err
}
if config.AdaptiveMinThreshold >= config.AdaptiveMaxThreshold {
return fmt.Errorf("adaptiveMinThreshold必须小于adaptiveMaxThreshold")
@@ -618,19 +681,8 @@ func CircuitBreakerMiddleware(r *ghttp.Request) {
return
}
cbInfoVal, ok := circuitBreakers.Load(serviceName)
if !ok {
r.Middleware.Next()
return
}
cbInfo, ok := cbInfoVal.(*CircuitBreakerInfo)
if !ok {
r.Middleware.Next()
return
}
config := cbInfo.Config
if !config.Enabled {
cbInfo, config := getCircuitBreakerInfoAndConfig(serviceName)
if cbInfo == nil || config == nil || !config.Enabled {
r.Middleware.Next()
return
}
@@ -718,16 +770,7 @@ func CircuitBreakerMiddleware(r *ghttp.Request) {
duration := time.Since(startTime)
// 记录响应时间统计
durationNs := duration.Nanoseconds()
cbInfo.Metrics.TotalResponseTime.Add(durationNs)
// 原子更新最小和最大响应时间
atomicUpdateMin(&cbInfo.Metrics.MinResponseTime, durationNs)
atomicUpdateMax(&cbInfo.Metrics.MaxResponseTime, durationNs)
if duration > config.SlowRequestThresholdParsed {
cbInfo.Metrics.SlowRequests.Add(1)
}
updateResponseTimeStats(cbInfo, duration, config)
isSuccess := isSuccessStatusCode(cbInfo, statusCode)
@@ -1036,23 +1079,10 @@ func CircuitBreakerHealthCheckHandler(r *ghttp.Request) {
halfOpenServices++
}
lastResetTime := cbInfo.Metrics.LastResetTime.Load()
var lastResetTimeStr string
if lastResetTime > 0 {
lastResetTimeStr = time.Unix(lastResetTime, 0).Format("2006-01-02 15:04:05")
}
lastOpenTime := cbInfo.Metrics.LastOpenTime.Load()
var lastOpenTimeStr string
if lastOpenTime > 0 {
lastOpenTimeStr = time.Unix(lastOpenTime, 0).Format("2006-01-02 15:04:05")
}
nextRetryTime := cbInfo.Metrics.NextRetryTime.Load()
var nextRetryTimeStr string
if nextRetryTime > 0 {
nextRetryTimeStr = time.Unix(nextRetryTime, 0).Format("2006-01-02 15:04:05")
}
// 格式化时间字符串
lastResetTimeStr := formatUnixTime(cbInfo.Metrics.LastResetTime.Load())
lastOpenTimeStr := formatUnixTime(cbInfo.Metrics.LastOpenTime.Load())
nextRetryTimeStr := formatUnixTime(cbInfo.Metrics.NextRetryTime.Load())
status[serviceName] = map[string]interface{}{
"resource": cbInfo.ResourceName,
@@ -1084,8 +1114,8 @@ func isAdminIP(r *ghttp.Request) bool {
return false
}
allowedIPs := getAllowedIPs()
allowedCIDRs := getAllowedCIDRs()
// 一次性获取IP和CIDR列表，减少锁操作
allowedIPs, allowedCIDRs := getAllowedIPsAndCIDRs()
// 如果没有任何限制，允许访问
if len(allowedIPs) == 0 && len(allowedCIDRs) == 0 {
@@ -1098,8 +1128,7 @@ func isAdminIP(r *ghttp.Request) bool {
}
// 检查CIDR白名单
clientNetIP := net.ParseIP(clientIP)
if clientNetIP != nil {
if clientNetIP := net.ParseIP(clientIP); clientNetIP != nil {
for _, cidrNet := range allowedCIDRs {
if cidrNet.Contains(clientNetIP) {
return true