Skip to content

Latest commit

 

History

History
373 lines (310 loc) · 12 KB

v5.0.md

File metadata and controls

373 lines (310 loc) · 12 KB

marpc v5.0 — 重试 & 熔断 & 灰度路由

版本概述

v5.0 目标是提升调用可靠性和流量管理能力:

  1. 重试机制:可配置重试次数、超时、自动切换节点
  2. 熔断器:滑动窗口统计失败率,状态机(Closed → Open → Half-Open → Closed)
  3. 灰度路由:按比例将流量路由到灰度节点,支持灰度发布

代码结构变更

marpc-core/src/main/java/com/malinghan/marpc/
├── retry/                           # v5.0 新增
│   └── RetryPolicy.java             # 重试策略配置
├── circuitbreaker/                  # v5.0 新增
│   ├── CircuitBreakerState.java     # 熔断器状态枚举
│   ├── CircuitBreakerConfig.java    # 熔断器配置
│   └── CircuitBreaker.java          # 熔断器实现
└── router/                          # v5.0 新增
    ├── Router.java                  # 路由接口
    └── GrayRouter.java              # 灰度路由实现

marpc-demo-consumer/src/main/java/com/malinghan/marpc/consumer/
├── Scene7Retry.java                 # 重试场景演示
├── Scene8CircuitBreaker.java        # 熔断器场景演示
└── Scene9GrayRouter.java            # 灰度路由场景演示

核心功能

1. 重试机制

配置项:

marpc:
  retry:
    maxRetries: 2                    # 最大重试次数(不含首次调用)
    timeout: 3000                    # 超时时间(毫秒)
    switchInstanceOnRetry: true      # 重试时是否切换节点

核心逻辑:

private RpcResponse invokeWithRetry(RpcRequest request) {
    Set<String> triedInstances = new HashSet<>();
    int attempts = 0;
    int maxAttempts = 1 + retryPolicy.getMaxRetries();

    while (attempts < maxAttempts) {
        attempts++;
        String instance = selectInstance(triedInstances);  // 切换节点
        triedInstances.add(instance);

        try {
            RpcResponse response = post(instance, request);
            if (response.isStatus()) {
                circuitBreaker.onSuccess();
                return response;
            }
            return response;  // 业务异常不重试
        } catch (MarpcNetworkException e) {
            circuitBreaker.onFailure();
            if (attempts < maxAttempts) {
                log.warn("[Retry] 第 {} 次调用失败: {}", attempts, instance);
            }
        }
    }
    throw new MarpcNetworkException(...);
}

特性:

  • 网络异常自动重试,业务异常不重试
  • 重试时自动切换节点(避免打到同一故障实例)
  • 配置超时时间,防止长时间阻塞
  • 与熔断器联动:成功调用重置失败计数,失败调用累计失败次数

2. 熔断器

配置项:

marpc:
  circuitbreaker:
    enabled: false                   # 是否启用熔断器
    faultLimit: 5                    # 触发熔断的失败次数阈值
    halfOpenInitialDelay: 10000      # 熔断后首次进入半开状态的延迟(毫秒)
    halfOpenDelay: 5000              # 半开状态下探测失败后再次进入半开的延迟(毫秒)
    windowSize: 10                   # 滑动窗口大小(秒)

状态机:

CLOSED(关闭)
    │
    │ 失败次数 >= faultLimit
    ▼
OPEN(打开)─────────────────────────┐
    │                                │
    │ 等待 halfOpenInitialDelay      │ 快速失败
    ▼                                │
HALF_OPEN(半开)                    │
    │                                │
    ├─ 探测成功 ──> CLOSED           │
    └─ 探测失败 ──> OPEN ────────────┘

核心逻辑:

public void preCall() {
    if (state.get() == CircuitBreakerState.OPEN) {
        long elapsed = System.currentTimeMillis() - openTime.get();
        if (elapsed >= config.getHalfOpenInitialDelay()) {
            state.compareAndSet(OPEN, HALF_OPEN);  // 进入半开状态
        } else {
            throw new MarpcFrameworkException(CIRCUIT_BREAKER_OPEN, "熔断器已打开");
        }
    }
}

public void onFailure() {
    int failures = failureCount.incrementAndGet();
    if (state.get() == CLOSED && failures >= config.getFaultLimit()) {
        state.compareAndSet(CLOSED, OPEN);  // 触发熔断
        openTime.set(System.currentTimeMillis());
    } else if (state.get() == HALF_OPEN) {
        state.compareAndSet(HALF_OPEN, OPEN);  // 探测失败,重新打开
    }
}

特性:

  • 滑动时间窗口统计失败率(超过窗口时间重置计数)
  • OPEN 状态快速失败,不发起远程调用
  • HALF_OPEN 状态允许探测请求,成功则自动恢复
  • 与重试机制联动:每次调用结果更新熔断器状态

3. 灰度路由

配置项:

marpc:
  router:
    gray:
      enabled: false                 # 是否启用灰度路由
      ratio: 50                      # 灰度流量比例(0-100)

核心逻辑:

@Override
public List<String> route(List<String> instances) {
    List<String> grayList = new ArrayList<>();
    List<String> normalList = new ArrayList<>();

    for (String instance : instances) {
        if (grayInstances.containsKey(instance)) {
            grayList.add(instance);
        } else {
            normalList.add(instance);
        }
    }

    // 按比例路由
    int dice = random.nextInt(100);
    if (dice < grayRatio) {
        return grayList;  // 命中灰度流量
    } else {
        return normalList;  // 命中正常流量
    }
}

使用方式:

// 标记灰度实例
grayRouter.markGray("localhost:8082");

// 取消灰度标记
grayRouter.unmarkGray("localhost:8082");

// 清空所有灰度标记
grayRouter.clearGray();

特性:

  • 按比例将流量路由到灰度节点(0-100%)
  • 支持动态标记/取消灰度实例
  • 无灰度实例时返回全部正常实例
  • 无正常实例时返回全部灰度实例

调用流程

Consumer 调用
    │
    ▼
┌─────────────────────────────────────────┐
│  preFilter(MockFilter → CacheFilter)   │
│  任意返回非 null → 短路                  │
└─────────────────────────────────────────┘
    │ 全部 miss
    ▼
┌─────────────────────────────────────────┐
│  熔断器检查(circuitBreaker.preCall)    │
│  OPEN 状态 → 快速失败                   │
│  HALF_OPEN 状态 → 允许探测              │
└─────────────────────────────────────────┘
    │ CLOSED 或 HALF_OPEN
    ▼
┌─────────────────────────────────────────┐
│  路由筛选(GrayRouter)                  │
│  按比例返回灰度/正常实例列表             │
└─────────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────────┐
│  负载均衡(RoundRobin / Random)         │
│  从筛选后的实例中选择一个                │
└─────────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────────┐
│  重试调用(invokeWithRetry)             │
│  失败自动重试,切换节点                  │
│  成功 → circuitBreaker.onSuccess()      │
│  失败 → circuitBreaker.onFailure()      │
└─────────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────────┐
│  postFilter(逆序执行)                  │
│  CacheFilter 写入缓存                   │
└─────────────────────────────────────────┘
    │
    ▼
  返回结果

Demo 测试场景

Scene7:重试机制

=== Scene7: 重试机制 ===
  提示:需要手动关闭 Provider 或模拟网络故障来观察重试效果
  配置:maxRetries=2, timeout=3000ms, switchInstanceOnRetry=true
  [Retry] 第 1 次调用失败: localhost:8080, 原因: call failed
  [Retry] 第 2 次调用失败: localhost:8081, 原因: call failed
  [Retry] 第 3 次调用失败: localhost:8082, 原因: call failed
  [FAIL] 调用失败: 调用失败,已重试 2 次
=== Scene7 完成 ===

Scene8:熔断器

=== Scene8: 熔断器 ===
  提示:需要启用熔断器配置(marpc.circuitbreaker.enabled=true)
  配置:faultLimit=5, halfOpenInitialDelay=10000ms
  当前熔断器状态: CLOSED, 失败计数: 0
  熔断器处于 CLOSED 状态,正常调用
  [SUCCESS] 调用成功: hello, circuit-test
=== Scene8 完成 ===

触发熔断后:

  当前熔断器状态: OPEN, 失败计数: 5
  [CircuitBreaker] 失败次数达到 5,进入 OPEN 状态
  熔断器已打开,调用将快速失败
  [EXPECTED] 熔断器拦截: 熔断器已打开,快速失败(剩余 8523ms)

Scene9:灰度路由

=== Scene9: 灰度路由 ===
  提示:需要启用灰度路由配置(marpc.router.gray.enabled=true)
  配置:grayRatio=50(50% 流量路由到灰度实例)
  已标记灰度实例: localhost:8082
  执行 10 次调用,观察灰度/正常实例分布:
    [GrayRouter] 命中灰度流量(50%),返回灰度实例: [localhost:8082]
    第 1 次调用成功: hello, gray-test-0
    [GrayRouter] 命中正常流量,返回正常实例: [localhost:8080, localhost:8081]
    第 2 次调用成功: hello, gray-test-1
    ...
  已清除灰度标记
=== Scene9 完成 ===

配置示例

完整配置(marpc-demo-consumer/application.yaml):

marpc:
  zk:
    address: localhost:2181
  app: marpc-app
  env: dev
  provider:
    instance: localhost:8080
  loadbalancer: roundrobin
  filter:
    cache:
      enabled: true
    mock:
      enabled: true
  retry:
    maxRetries: 2
    timeout: 3000
    switchInstanceOnRetry: true
  circuitbreaker:
    enabled: false
    faultLimit: 5
    halfOpenInitialDelay: 10000
    halfOpenDelay: 5000
    windowSize: 10
  router:
    gray:
      enabled: false
      ratio: 50

v4.0 → v5.0 升级要点

变更点 v4.0 v5.0
重试机制 无 RetryPolicy,可配置重试次数、超时、切换节点
熔断器 无 CircuitBreaker,状态机 + 滑动窗口
灰度路由 无 GrayRouter,按比例路由到灰度节点
调用流程 Filter → 远程调用 Filter → 熔断器 → 路由 → 负载均衡 → 重试调用
MarpcInvocationHandler 直接调用 invokeWithRetry + 熔断器联动
ConsumerBootstrap 无路由 接收 List<Router>,路由筛选后再负载均衡
MarpcConfig 无重试/熔断/路由 RetryPolicy、CircuitBreaker、GrayRouter Bean
Consumer 配置 Filter 开关 新增 retry.*、circuitbreaker.*、router.gray.*

后续规划(v6.0)

  • Netty 传输层:OkHttp → Netty 长连接,自定义二进制协议
  • 连接池管理:复用连接,降低连接建立开销
  • 协议优化:替换 HTTP+JSON,减少序列化开销
  • 监控指标:调用耗时、成功率、熔断次数、灰度流量分布
  • 配置中心:动态调整重试次数、熔断阈值、灰度比例