Kitex服务降级

2024-01-17T14:21:02+08:00 | 6分钟阅读 | 更新于 2024-01-17T14:21:02+08:00

@

概述

服务降级（Fallback / Degradation） 是指当服务依赖的下游出现故障、超时或熔断器打开时，不再执行正常的业务逻辑，而是返回一个预设的兜底结果，以保证核心链路的可用性。

降级的核心思想：宁可返回不完美的结果，也不要让用户看到错误。

与熔断的区别：

  • 熔断关注的是"切断调用，防止故障扩散"
  • 降级关注的是"返回兜底结果，保证用户体验"

两者通常配合使用：熔断是手段，降级是目的。

请求进入
  ↓
┌──────────────┐
│  熔断检查     │── 熔断打开 ──→ 进入降级逻辑
└──────┬───────┘
       ↓ 允许
┌──────────────┐
│  正常调用     │── 成功 ──→ 返回正常结果
└──────┬───────┘
       ↓ 失败/超时
┌──────────────┐
│  降级逻辑     │── 返回兜底数据
└──────────────┘

降级策略分类

策略说明适用场景
默认值降级返回固定的默认值或空值列表查询、配置读取
缓存降级返回本地/远程缓存中的旧数据商品详情、用户信息
Mock 降级返回模拟数据非核心功能、开发测试环境
页面降级返回简化版页面或静态页前端展示类服务
组合降级多级降级：实时→缓存→默认值高可用要求高的核心链路

方案一：Middleware 实现通用降级

在 Kitex 客户端通过 Middleware 统一拦截，实现声明式降级。

降级处理器接口

package fallback

import (
	"context"
)

// FallbackFunc 降级函数签名
type FallbackFunc func(ctx context.Context, req, err error) (resp interface{}, fallbackErr error)

// FallbackProvider 降级提供者
type FallbackProvider struct {
	funcMap map[string]FallbackFunc // 按方法名映射降级函数
}

// NewFallbackProvider 创建降级提供者
func NewFallbackProvider() *FallbackProvider {
	return &FallbackProvider{
		funcMap: make(map[string]FallbackFunc),
	}
}

// Register 注册特定方法的降级逻辑
func (fp *FallbackProvider) Register(method string, fn FallbackFunc) {
	fp.funcMap[method] = fn
}

// Get 获取指定方法的降级函数
func (fp *FallbackProvider) Get(method string) FallbackFunc {
	return fp.funcMap[method]
}

降级 Middleware

func FallbackMW(fp *FallbackProvider) endpoint.Middleware {
	return func(next endpoint.Endpoint) endpoint.Endpoint {
		return func(ctx context.Context, req, resp interface{}) (err error) {
			// 1. 执行正常调用
			err = next(ctx, req, resp)

			// 2. 如果调用成功，直接返回
			if err == nil {
				return nil
			}

			// 3. 提取方法名
			method := getMethodName(ctx)

			// 4. 查找是否有对应的降级逻辑
			fallbackFn := fp.Get(method)
			if fallbackFn == nil {
				// 没有降级逻辑，向上返回错误
				return err
			}

			// 5. 执行降级
			fallbackResp, fallbackErr := fallbackFn(ctx, req, err)
			if fallbackErr != nil {
				// 降级也失败了，返回原始错误
				return err
			}

			// 6. 将降级结果写回 resp（需要类型断言）
			if respVal, ok := resp.(interface{ Reset() }); ok {
				respVal.Reset()
			}
			// 具体写入方式取决于 RPC 框架的 resp 结构
			_ = fallbackResp

			return nil
		}
	}
}

使用示例

// 1. 创建降级提供者
fallbackProvider := NewFallbackProvider()

// 2. 注册降级逻辑
fallbackProvider.Register("GetUserInfo", func(ctx context.Context, req interface{}, err error) (interface{}, error) {
	// 从本地缓存读取用户信息
	cacheKey := req.(*GetUserInfoRequest).UserId
	userInfo := getUserInfoFromCache(cacheKey)
	if userInfo != nil {
		return &GetUserInfoResponse{
			User: userInfo,
			Fallback: true, // 标记为降级数据
		}, nil
	}
	// 缓存也没有，返回默认值
	return &GetUserInfoResponse{
		User: &UserInfo{
			Name: "未知用户",
		},
		Fallback: true,
	}, nil
})

fallbackProvider.Register("QueryProductList", func(ctx context.Context, req interface{}, err error) (interface{}, error) {
	// 返回缓存的商品列表
	list := getProductListFromCache()
	return &QueryProductListResponse{
		Products: list,
		Fallback: true,
	}, nil
})

// 3. 注入客户端
client, err := shop.NewClient(
	"dqq.shop",
	client.WithMiddleware(FallbackMW(fallbackProvider)),
	client.WithMiddleware(CircuitBreakerMW(cb)), // 与熔断配合
	client.WithRPCTimeout(200*time.Millisecond),
)

方案二：熔断 + 降级联动

这是最常见的生产实践：熔断器打开时自动走降级逻辑。

func CircuitBreakerWithFallback(cb *CircuitBreaker, fp *FallbackProvider) endpoint.Middleware {
	return func(next endpoint.Endpoint) endpoint.Endpoint {
		return func(ctx context.Context, req, resp interface{}) (err error) {
			// 1. 熔断检查
			if !cb.AllowRequest() {
				// 熔断打开，走降级
				method := getMethodName(ctx)
				fallbackFn := fp.Get(method)
				if fallbackFn != nil {
					fallbackResp, _ := fallbackFn(ctx, req, ErrCircuitBreakerOpen)
					if respVal, ok := resp.(interface{ Reset() }); ok {
						respVal.Reset()
					}
					_ = fallbackResp
					return nil
				}
				// 无降级逻辑，返回熔断错误
				return ErrCircuitBreakerOpen
			}

			// 2. 正常调用
			err = next(ctx, req, resp)
			if err != nil {
				cb.RecordFailure()
			} else {
				cb.RecordSuccess()
			}

			// 3. 调用失败且存在降级逻辑
			if err != nil {
				method := getMethodName(ctx)
				fallbackFn := fp.Get(method)
				if fallbackFn != nil {
					fallbackResp, fallbackErr := fallbackFn(ctx, req, err)
					if fallbackErr == nil {
						if respVal, ok := resp.(interface{ Reset() }); ok {
							respVal.Reset()
						}
						_ = fallbackResp
						return nil
					}
				}
			}

			return err
		}
	}
}

方案三：Sentinel 降级规则

如果使用 Apache Sentinel，可以通过规则配置实现声明式降级：

import (
	"github.com/alibaba/sentinel-golang/core/degrade"
)

// 降级规则：当错误率达到阈值时触发降级
rule := degrade.Rule{
	Resource:              "dqq.shop/Service.GetUserInfo",
	Strategy:              degrade.StrategyErrorRatio,
	Threshold:             0.5,           // 错误率 50%
	MinRequestNumber:      10,            // 最少 10 个请求
	StatIntervalMs:        10000,         // 统计窗口 10s
	RestoreTimeoutMs:      30000,         // 降级恢复时间 30s
}
degrade.PutRule(rule)

// Sentinel 支持三种降级策略：
// StrategyExceptionCount  — 异常数阈值
// StrategyErrorRatio      — 异常比率阈值
// StrategySlowRequestRatio — 慢调用比率阈值

Sentinel 配合 FallbackFunction 可以实现更细粒度的降级控制：

// Sentinel 的 FlowProtection 支持 fallback 回调
sentinel.SeatProtection("resource_name",
	sentinel.WithFallback(func(ctx context.Context, err error) {
		// 降级逻辑
		return fallbackResponse, nil
	}),
)

方案四：本地缓存降级（实战常用）

对于读多写少的场景，本地缓存是最实用的降级手段：

type CachedClient struct {
	inner    Client            // 真实的 Kitex 客户端
	cache    *localcache.Cache // 本地缓存
	ttl      time.Duration
	stats    *FallbackStats    // 降级统计
}

func (cc *CachedClient) GetUserInfo(ctx context.Context, req *GetUserInfoRequest) (*GetUserInfoResponse, error) {
	// 1. 先查本地缓存
	if cached := cc.cache.Get(req.UserId); cached != nil {
		cc.stats.IncCacheHit()
		resp := cached.(*GetUserInfoResponse)
		resp.Fallback = true
		return resp, nil
	}

	// 2. 缓存未命中，调用远程服务
	resp, err := cc.inner.GetUserInfo(ctx, req)
	if err != nil {
		cc.stats.IncFallback()
		// 3. 远程调用失败，返回缓存的旧数据（如果有的话）
		if cached := cc.cache.Get(req.UserId); cached != nil {
			resp := cached.(*GetUserInfoResponse)
			resp.Fallback = true
			return resp, nil
		}
		return nil, err
	}

	// 4. 成功则写入缓存
	cc.cache.Set(req.UserId, resp, cc.ttl)
	return resp, nil
}

降级标记与可观测性

降级数据应当与正常数据区分，以便排查问题和监控：

响应中标记降级

type BaseResponse struct {
	Fallback bool `json:"fallback,omitempty"` // 是否为降级数据
	FallbackReason string `json:"fallback_reason,omitempty"` // 降级原因
}

// 在降级函数中设置
return &GetUserInfoResponse{
	BaseResponse: BaseResponse{
		Fallback:       true,
		FallbackReason: "downstream_timeout",
	},
	User: userInfo,
}, nil

降级指标上报

type FallbackStats struct {
	totalRequests  prometheus.Counter
	fallbackCalls  prometheus.Counter
	cacheHits      prometheus.Counter
	fallbackErrors prometheus.Counter
}

func (fs *FallbackStats) IncFallback() {
	fs.fallbackCalls.Inc()
}

func (fs *FallbackStats) IncCacheHit() {
	fs.cacheHits.Inc()
}

// 在 Middleware 中集成
func FallbackWithMetricsMW(fp *FallbackProvider, stats *FallbackStats) endpoint.Middleware {
	return func(next endpoint.Endpoint) endpoint.Endpoint {
		return func(ctx context.Context, req, resp interface{}) (err error) {
			err = next(ctx, req, resp)
			if err != nil {
				method := getMethodName(ctx)
				fallbackFn := fp.Get(method)
				if fallbackFn != nil {
					stats.IncFallback()
					_, fallbackErr := fallbackFn(ctx, req, err)
					if fallbackErr != nil {
						stats.IncFallbackErrors()
					}
				}
			}
			return err
		}
	}
}

降级 vs 熔断 vs 限流

机制位置触发条件行为目的
限流服务端QPS/并发超限拒绝新请求保护自身
熔断客户端下游错误率过高切断调用，快速失败防止故障扩散
降级客户端熔断打开 / 调用失败返回兜底结果保证用户体验
三者协作关系：

限流（服务端）          熔断（客户端）         降级（客户端）
┌──────────┐         ┌──────────┐          ┌──────────┐
│ 保护自身  │         │ 切断调用  │          │ 兜底响应  │
│ 拒绝请求  │──────→  │ 快速失败  │──────→   │ 返回默认  │
│          │         │          │          │  数据     │
└──────────┘         └──────────┘          └──────────┘
   第一道防线           第二道防线             最后一道防线

最佳实践

1. 分级降级策略

按业务重要性分层设计降级方案：

级别策略示例
P0 核心实时 → 缓存 → 默认值 → 友好提示用户登录、支付
P1 重要实时 → 缓存 → 默认值商品列表、订单查询
P2 一般实时 → 缓存 → 跳过推荐、评论
P3 边缘直接跳过消息通知、日志上报

2. 降级数据时效性标注

所有降级数据必须标注来源和时间，避免误导用户：

type UserInfoResponse struct {
	User       *UserInfo `json:"user"`
	Fallback   bool      `json:"fallback"`
	CacheTime  string    `json:"cache_time,omitempty"`  // 缓存时间
	DataAge    string    `json:"data_age,omitempty"`    // 数据距今多久
}

3. 降级开关

通过配置中心控制降级开关，便于紧急场景手动触发：

var fallbackEnabled atomic.Bool

func init() {
	fallbackEnabled.Store(true) // 默认开启
	// 监听配置中心变更
	config.Watch("fallback.enabled", func(val bool) {
		fallbackEnabled.Store(val)
	})
}

func FallbackMW(fp *FallbackProvider) endpoint.Middleware {
	return func(next endpoint.Endpoint) endpoint.Endpoint {
		return func(ctx context.Context, req, resp interface{}) (err error) {
			err = next(ctx, req, resp)
			if err == nil {
				return nil
			}
			if !fallbackEnabled.Load() {
				return err // 降级关闭，直接返回错误
			}
			// ... 降级逻辑
		}
	}
}

4. 避免降级雪崩

  • 降级逻辑本身也要加超时和限流，避免降级函数阻塞
  • 降级返回的数据应尽可能轻量，避免二次调用重型依赖
  • 定期评估降级覆盖率，确保核心链路都有兜底方案

5. 灰度降级

对部分用户开启降级，观察效果后再全量：

func FallbackWithGrayMW(fp *FallbackProvider, grayRate float64) endpoint.Middleware {
	return func(next endpoint.Endpoint) endpoint.Endpoint {
		return func(ctx context.Context, req, resp interface{}) (err error) {
			err = next(ctx, req, resp)
			if err == nil {
				return nil
			}
			// 按用户 ID 哈希决定是否走降级
			userID := extractUserID(req)
			if hash(userID)%100 >= int(grayRate*100) {
				return err // 不走降级，返回真实错误
			}
			// 走降级逻辑
			fallbackFn := fp.Get(getMethodName(ctx))
			if fallbackFn != nil {
				fallbackResp, _ := fallbackFn(ctx, req, err)
				_ = fallbackResp
			}
			return nil
		}
	}
}

注意事项

  1. 降级不是万能药：降级数据可能与实时数据不一致，需告知用户
  2. 降级逻辑要简单：避免在降级函数中调用其他远程服务，否则可能引发连锁故障
  3. 注意缓存穿透：降级返回的缓存数据可能是空的，需设置合理的 TTL
  4. 监控覆盖率：统计降级触发次数和比例，及时发现异常
  5. 测试降级场景：通过混沌工程模拟下游故障，验证降级是否生效

相关笔记

  • [[Kitex/熔断]] — 熔断是降级的前置条件，两者通常配合使用
  • [[Kitex/限流]] — 限流保护服务端，降级保护客户端体验
  • [[Kitex/超时]] — 超时会触发降级逻辑
  • [[Kitex/中间件]] — 降级通过 Middleware 机制嵌入调用链
  • [[Kitex/重试]] — 重试失败后再走降级
上一页 Kitex处理Panic
下一页 Kitex日志
About Me

没什么想介绍的，一个很大众的码农…

喜欢代码，车，马，真的是 🐎

讨厌别人让我给自己的代码写注释 最厌烦别人的程序没有写注释

目标

学AI，加油！加油！