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 11:08:55 +08:00
parent 456bee2cae
commit b1f52bb328
1 changed files with 118 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

144
middleware/circuit_breaker.go
View File

@@ -37,6 +37,7 @@ type CircuitBreakerConfig struct {
FallbackMessage string // 降级提示消息
RequestTimeout int // 请求超时时间（毫秒）
DistributedTTL int // 分布式熔断状态TTL（秒）
AdminIPs []string // 允许重置熔断器的管理员IP列表
}
// CircuitBreakerMetrics 熔断器指标
@@ -46,6 +47,7 @@ type CircuitBreakerMetrics struct {
BlockRequests atomic.Int64 // 阻塞请求数
FailureRequests atomic.Int64 // 失败请求数
OpenCount atomic.Int64 // 熔断开启次数
LastResetTime atomic.Int64 // 上次重置时间（Unix时间戳）
}
// CircuitBreakerInfo 熔断器信息
@@ -158,6 +160,7 @@ func loadServiceCircuitBreakerConfig(serviceName string) *CircuitBreakerConfig {
fallbackMessage := g.Cfg().MustGet(ctx, key+".fallbackMessage", "").String()
requestTimeout := g.Cfg().MustGet(ctx, key+".requestTimeout", 30000).Int()
distributedTTL := g.Cfg().MustGet(ctx, key+".distributedTTL", 300).Int()
adminIPs := g.Cfg().MustGet(ctx, key+".adminIPs", "").String()
// 解析成功状态码
successCodes := g.Cfg().MustGet(ctx, key+".successStatusCodes", "200,201,204").String()
@@ -176,6 +179,9 @@ func loadServiceCircuitBreakerConfig(serviceName string) *CircuitBreakerConfig {
g.Log().Warningf(ctx, "服务 %s 的 slowRequestThreshold 解析失败，使用默认值: %v", serviceName, err)
}
// 解析管理员IP列表
adminIPList := parseAdminIPs(adminIPs)
return &CircuitBreakerConfig{
Enabled: enabled,
MaxFailures: maxFailures,
@@ -190,6 +196,7 @@ func loadServiceCircuitBreakerConfig(serviceName string) *CircuitBreakerConfig {
FallbackMessage: fallbackMessage,
RequestTimeout: requestTimeout,
DistributedTTL: distributedTTL,
AdminIPs: adminIPList,
}
}
@@ -206,6 +213,22 @@ func parseStatusCodes(str string) []int {
return codes
}
// parseAdminIPs 解析管理员IP列表
func parseAdminIPs(str string) []string {
if str == "" {
return nil
}
parts := strings.Split(str, ",")
ips := make([]string, 0, len(parts))
for _, part := range parts {
ip := strings.TrimSpace(part)
if ip != "" {
ips = append(ips, ip)
}
}
return ips
}
// filterServiceNames 过滤服务名（排除非服务配置的key）
func filterServiceNames(services map[string]interface{}) []string {
excludeKeys := map[string]bool{
@@ -282,6 +305,7 @@ func initServiceCircuitBreaker(serviceName string, config *CircuitBreakerConfig)
Config: config,
Metrics: &CircuitBreakerMetrics{},
}
cbInfo.Metrics.LastResetTime.Store(time.Now().Unix())
circuitBreakers.Store(serviceName, cbInfo)
strategy := "error_count"
@@ -300,15 +324,13 @@ func CircuitBreakerMiddleware(r *ghttp.Request) {
startTime := time.Now()
ctx := r.GetCtx()
// 从URL路径提取服务名
pathParts := strings.Split(strings.Trim(r.URL.Path, "/"), "/")
if len(pathParts) == 0 {
// 从URL路径提取服务名（改进提取逻辑）
serviceName := extractServiceName(r.URL.Path)
if serviceName == "" {
r.Middleware.Next()
return
}
serviceName := pathParts[0]
// 获取熔断器配置
val, ok := circuitBreakerConfigs.Load(serviceName)
if !ok {
@@ -391,7 +413,8 @@ func CircuitBreakerMiddleware(r *ghttp.Request) {
statusCode := r.Response.Status
duration := time.Since(startTime)
if !isSuccessStatusCode(resourceName, statusCode) {
// 使用提前获取的config判断状态码（性能优化）
if !isSuccessStatusCode(config, statusCode) {
// 记录异常
cbInfo.Metrics.FailureRequests.Add(1)
api.TraceError(entry, fmt.Errorf("request failed with status: %d", statusCode))
@@ -431,32 +454,39 @@ func sendFallbackResponse(r *ghttp.Request, serviceName string, config *CircuitB
}
// isSuccessStatusCode 判断HTTP状态码是否成功
func isSuccessStatusCode(resourceName string, statusCode int) bool {
serviceName := strings.TrimPrefix(resourceName, "service:")
if serviceName == "" {
// 默认只认为2xx是成功
return statusCode >= 200 && statusCode < 300
}
// 从配置中获取成功状态码列表
var serviceConfig *CircuitBreakerConfig
if val, ok := circuitBreakerConfigs.Load(serviceName); ok {
serviceConfig = val.(*CircuitBreakerConfig)
}
if serviceConfig != nil && len(serviceConfig.SuccessStatusCodes) > 0 {
for _, code := range serviceConfig.SuccessStatusCodes {
func isSuccessStatusCode(config *CircuitBreakerConfig, statusCode int) bool {
if len(config.SuccessStatusCodes) > 0 {
for _, code := range config.SuccessStatusCodes {
if statusCode == code {
return true
}
}
return false
}
// 默认：2xx状态码为成功
return statusCode >= 200 && statusCode < 300
}
// extractServiceName 从URL路径提取服务名（改进提取逻辑）
func extractServiceName(path string) string {
// 去除首尾斜杠并分割
path = strings.Trim(path, "/")
if path == "" {
return ""
}
parts := strings.Split(path, "/")
if len(parts) == 0 {
return ""
}
serviceName := parts[0]
// 验证服务名是否在已配置的熔断器中
if _, ok := circuitBreakerConfigs.Load(serviceName); ok {
return serviceName
}
return ""
}
// isCircuitBreakerOpenInDistributed 检查分布式熔断状态
func isCircuitBreakerOpenInDistributed(ctx context.Context, resourceName string) bool {
key := fmt.Sprintf("circuit_breaker:%s:state", resourceName)
@@ -536,10 +566,16 @@ func registerStateChangeListeners() {
return
}
// 注册默认监听器
// 注册默认监听器（区分日志级别）
RegisterStateChangeListener("default", func(serviceName string, fromState, toState CircuitBreakerState) {
g.Log().Infof(context.Background(), "熔断器状态变化: service=%s, %s -> %s",
serviceName, fromState, toState)
// Open状态使用Warning级别，Closed状态使用Info级别
if toState == StateOpen {
g.Log().Warningf(context.Background(), "熔断器状态变化: service=%s, %s -> %s",
serviceName, fromState, toState)
} else {
g.Log().Infof(context.Background(), "熔断器状态变化: service=%s, %s -> %s",
serviceName, fromState, toState)
}
})
}
@@ -585,6 +621,12 @@ func CircuitBreakerHealthCheckHandler(r *ghttp.Request) {
}
// 从Metrics中读取数据（修复数据准确性问题）
lastResetTime := cbInfo.Metrics.LastResetTime.Load()
var lastResetTimeStr string
if lastResetTime > 0 {
lastResetTimeStr = time.Unix(lastResetTime, 0).Format("2006-01-02 15:04:05")
}
status[serviceName] = map[string]interface{}{
"resource": cbInfo.ResourceName,
"state": string(cbInfo.State),
@@ -595,6 +637,7 @@ func CircuitBreakerHealthCheckHandler(r *ghttp.Request) {
"blockRequests": cbInfo.Metrics.BlockRequests.Load(),
"failureRequests": cbInfo.Metrics.FailureRequests.Load(),
"openCount": cbInfo.Metrics.OpenCount.Load(),
"lastResetTime": lastResetTimeStr,
}
cbInfo.mu.RUnlock()
@@ -617,6 +660,39 @@ func CircuitBreakerHealthCheckHandler(r *ghttp.Request) {
})
}
// isAdminIP 检查请求IP是否在管理员白名单中
func isAdminIP(r *ghttp.Request) bool {
clientIP := r.GetClientIp()
if clientIP == "" {
return false
}
// 检查所有服务的adminIPs配置
var allowedIPs []string
circuitBreakerConfigs.Range(func(key, value interface{}) bool {
config := value.(*CircuitBreakerConfig)
if len(config.AdminIPs) > 0 {
allowedIPs = append(allowedIPs, config.AdminIPs...)
}
return true
})
// 如果没有配置白名单，允许所有IP（向后兼容）
if len(allowedIPs) == 0 {
return true
}
// 检查IP是否在白名单中
for _, allowedIP := range allowedIPs {
if clientIP == allowedIP {
return true
}
}
g.Log().Warningf(r.GetCtx(), "熔断器重置请求被拒绝，IP不在白名单中: %s", clientIP)
return false
}
// CircuitBreakerResetHandler 熔断器手动重置接口（仅限管理后台调用）
func CircuitBreakerResetHandler(r *ghttp.Request) {
serviceName := r.Get("service").String()
@@ -628,6 +704,15 @@ func CircuitBreakerResetHandler(r *ghttp.Request) {
return
}
// 权限验证：检查IP是否在白名单中
if !isAdminIP(r) {
r.Response.WriteJsonExit(ghttp.DefaultHandlerResponse{
Code: 403,
Message: "权限不足，禁止访问",
})
return
}
resourceName := fmt.Sprintf("service:%s", serviceName)
// 获取当前服务的所有规则
@@ -658,13 +743,20 @@ func CircuitBreakerResetHandler(r *ghttp.Request) {
}
}
// 更新内存状态
// 更新内存状态并重置指标
if val, ok := circuitBreakers.Load(serviceName); ok {
cbInfo := val.(*CircuitBreakerInfo)
cbInfo.mu.Lock()
cbInfo.State = StateClosed
cbInfo.LastOpenTime = time.Time{}
cbInfo.NextRetryTime = time.Time{}
// 重置指标
cbInfo.Metrics.TotalRequests.Store(0)
cbInfo.Metrics.PassRequests.Store(0)
cbInfo.Metrics.BlockRequests.Store(0)
cbInfo.Metrics.FailureRequests.Store(0)
cbInfo.Metrics.OpenCount.Store(0)
cbInfo.Metrics.LastResetTime.Store(time.Now().Unix())
cbInfo.mu.Unlock()
}