熔断器：自动检测故障服务，快速失败保护系统

Created By RV, and licensed with Creative Commons "[CC BY-NC-ND 4.0](https://creativecommons.org/licenses/by-nc-nd/4.0/"

目的与场景

在分布式系统中，一个下游服务的故障可能引发上游连锁反应——调用方不断重试、线程阻塞、资源耗尽，最终拖垮整个集群。

在这一章，讲解如何使用 Kree4X 内置的熔断器（Circuit Breaker），在检测到服务持续故障时自动切断调用，快速失败以保护系统，并在服务恢复后自动恢复调用。

一. 概念

1. 什么是熔断器

熔断器是一个状态机，守护在调用方与被调用服务之间，监控每一次调用的成功与失败。

它有三个状态：

• Closed（关闭）：正常放行所有调用。持续记录调用结果，当失败率或慢调用率达到阈值时，切换到 Open。
• Open（打开）：拒绝所有调用，直接抛出 CircuitBreakerError（code=503）。经过设定的冷却时间后，自动切换到 Half-Open。
• Half-Open（半开）：允许有限数量的探测调用通过。探测成功则回到 Closed，探测失败则重新回到 Open。

  ┌──────────┐   失败率达阈值   ┌──────────┐   冷却时间到期   ┌───────────┐
  │  Closed  │ ──────────────→ │   Open   │ ──────────────→ │ Half-Open │
  │ （放行） │                  │ （拒绝） │                  │ （探测）  │
  └──────────┘ ←────────────── └──────────┘ ←────────────── └───────────┘
                   探测失败                   探测成功

2. 为什么需要熔断器

当服务持续不可用时，调用方不断发起请求只会：

• 浪费网络带宽和计算资源
• 增加下游恢复的负担
• 可能耗尽调用方的连接池或线程池，引发自身故障

熔断器的作用是快速失败——在检测到故障后立即拒绝调用，让调用方有时间处理降级逻辑，同时给下游服务恢复的空间。

3. 默认关闭，按需开启

Kree4X 的熔断器默认关闭——创建节点后不会自动拦截调用。

需要显式调用 enableCircuitBreaker(kreex) 开启：

import { create, enableCircuitBreaker } from '@kree4js/kree4n'

const node = create('my-node')
enableCircuitBreaker(node)  // 开启熔断器

关闭熔断器（恢复直通）：

node.circuitBreaker.disable()

4. 阈值：BreakerThresholds

熔断器根据滑动窗口内的调用统计做决策，阈值通过 BreakerThresholds 配置：

5. 慢调用检测（可选）

除了错误率，熔断器还可以检测慢调用：

• slowCallTime：一次调用耗时超过此值（ms）即判定为慢调用
• slowCallRate：慢调用占比达到此阈值时触发熔断

两者均未设置时，慢调用检测关闭。两者同时设置时，任一条件满足即触发熔断（OR 语义）。

// 慢调用检测：耗时超过 200ms 算慢调用，慢调用占比达 50% 触发熔断
node.circuitBreaker.addOrUpdateRule(
  new Rule(
    { nodeId: 'target-node', service: 'svc' },
    { minSamples: 5, slowCallTime: 200, slowCallRate: 0.5 }
  )
)

6. 桶策略：决定分桶粒度

熔断器为每一个 (service, nodeId, method) 组合维护一个独立的breaker实例。桶策略（BarrelPolicy）决定统计的聚合粒度。

四种内置策略：

全局设置桶策略：

const { ShareServiceMethodPolicy } = Kree4JS.CircuitBreaker
node.circuitBreaker.breakerBarrel(new ShareServiceMethodPolicy())

集群级设置（仅对该服务存根生效）：

const svc = node.service('my-svc')
svc.breakerBarrel(new ShareServicePolicy())

7. 规则：Rule

规则（Rule）将路由键与阈值绑定。路由键决定规则作用的范围：

规则支持热更新——修改阈值后立即生效，无需重启。

8. 恢复机制

Open 状态的 breaker 不会永久拒绝调用。内部的 TimeWheel 在 openDwellTime 到期后自动将其翻转为 Half-Open，允许探测调用通过。

探测调用的成功或失败决定最终走向：

• 探测成功 → Closed（恢复正常）
• 探测失败 → Open（重新冷却）

二. 示例代码

在下边的示例中，我们模拟一个持续故障的服务：

• node-a：注册 svc 服务，hello 方法始终抛错（模拟宕机）
• node-b：调用方，开启熔断器，自定义规则触发熔断

1. 搭建拓扑

import Kree4JS from '@kree4js/kree4js'
import { create, enableCircuitBreaker } from '@kree4js/kree4n'

const { Rule } = Kree4JS.CircuitBreaker

const nodeA = create('node-a', 'Flaky provider')
nodeA.register('svc', {
  hello () { throw new Error('service unavailable') }
})
nodeA.listen('tcp://127.0.0.1:8070', { frameLimit: 512 })

const nodeB = create('node-b', 'CB caller')
nodeB.attach('tcp://127.0.0.1:8070', { frameLimit: 512 })

await nodeA.start()
await nodeB.start()
enableCircuitBreaker(nodeB)

2. 自定义规则：快速触发

默认阈值需要 10 次失败才触发，这里用低 minSamples 演示：

nodeB.circuitBreaker.addOrUpdateRule(
  new Rule(
    { nodeId: 'node-a-id', service: 'svc' },
    { minSamples: 2, errorRate: 0.5, openDwellTime: 5000 }
  )
)

3. 连续调用触发熔断

const svc = nodeB.service('svc').timeout(5000)

// 第1次调用失败：minSamples=2 未达，breaker 仍 closed
try { await svc.hello('a') } catch (_) {}
console.log(nodeB.circuitBreaker.getState({ nodeId: 'node-a-id', service: 'svc' }))
// → 'closed'

// 第2次调用失败：minSamples=2 达到，errorRate=1.0 >= 0.5，breaker → open
try { await svc.hello('b') } catch (_) {}
console.log(nodeB.circuitBreaker.getState({ nodeId: 'node-a-id', service: 'svc' }))
// → 'open'

4. 熔断中：调用被快速失败

try {
  await svc.hello('c')
} catch (e) {
  console.log(e.name)  // → 'CircuitBreakerError'
  console.log(e.code)  // → 503
}

5. 状态观察

// 实时指标
const metrics = nodeB.circuitBreaker.getMetrics({ nodeId: 'node-a-id', service: 'svc' })
console.log(metrics.callCount)   // 窗口内总调用数
console.log(metrics.errorRate)   // 当前错误率

// 当前阈值
const thresholds = nodeB.circuitBreaker.getThresholds({ nodeId: 'node-a-id', service: 'svc' })
console.log(thresholds.errorRate)      // 0.5
console.log(thresholds.minSamples)     // 2
console.log(thresholds.openDwellTime)  // 5000

6. 恢复演示

// 等待 5s（openDwellTime），breaker 自动进入 half-open
await new Promise(resolve => setTimeout(resolve, 5100))

console.log(nodeB.circuitBreaker.getState({ nodeId: 'node-a-id', service: 'svc' }))
// → 'half-open'

// Half-Open 探测调用：成功则 closed，失败则重新 open
try { await svc.hello('d') } catch (_) {}
console.log(nodeB.circuitBreaker.getState({ nodeId: 'node-a-id', service: 'svc' }))
// → 'open'（探测失败，重新冷却）

7. 强制操作

// 强制关闭：恢复放行
nodeB.circuitBreaker.forceClose({ nodeId: 'node-a-id', service: 'svc' })

// 强制打开：立即拒绝
nodeB.circuitBreaker.forceOpen({ nodeId: 'node-a-id', service: 'svc' })

// 强制重置：metrics 归零，状态回到 closed
nodeB.circuitBreaker.forceReset({ nodeId: 'node-a-id', service: 'svc' })

8. 清除规则

nodeB.circuitBreaker.clearRules()
// 所有 breaker 恢复默认阈值，状态回到 closed

三. 须强调的细节

1. 默认关闭，enableCircuitBreaker 才生效

create() 创建的节点，熔断器默认关闭。

不调用 enableCircuitBreaker(kreex) 或 kreex.circuitBreaker.enable()，所有调用直通，不经过熔断器拦截。

enableCircuitBreaker 是幂等的，多次调用安全。

2. Breaker 懒创建

Breaker 不是在 addOrUpdateRule 时创建的，而是在首次调用时按桶策略分配。

未调用过的服务不会有 breaker 实例，getState 返回 undefined。

3. minSamples 未达到前不触发

滑动窗口内的调用次数不足 minSamples 时，即使错误率 100% 也不会触发熔断。

这避免了样本过少导致的误判——例如首次调用失败就熔断是不合理的。

4. HALF-OPEN 的 probe slot 机制

Half-Open 状态下，允许 halfOpenSlots 个并发探测调用通过（默认 1 个）。

探测调用成功 → Closed；失败 → 重新 Open。

探测调用数量有上限，避免恢复期间大量流量涌入。

5. 四种桶策略差异

桶策略决定了 breaker 的分桶粒度，直接影响故障隔离的范围：

• PerServicePolicy（默认）：每个 (service, nodeId) 独立。node-a 的 svc 故障不影响 node-b 的 svc。
• PerNodePolicy：每个 nodeId 共享。node-a 的任一服务故障会影响该节点的所有服务。
• ShareServicePolicy：同名服务跨节点共享。任一节点故障会熔断所有节点的同名服务。
• ShareServiceMethodPolicy：每个 (service, method) 独立。同一服务的 hello 故障不影响 bye。

6. 节点下线自动清理

当节点从 grid 中移除时，该节点的 breaker 实例和关联的 rule 会被自动清理，不会残留。

四. 涉及到的API

1. 开启/关闭熔断器

/**
 * 开启熔断器（幂等）。
 *
 * @param {KreeX} kreex - KreeX 实例。
 * @returns {KreeX} 同一实例，支持链式。
 */
enableCircuitBreaker(kreex): KreeX

// 等价于
kreex.circuitBreaker.enable()

/**
 * 关闭熔断器：拦截器卸载，已有 breaker 清除。
 */
kreex.circuitBreaker.disable()

2. 规则管理

/**
 * 添加或更新规则（热生效）。
 */
kreex.circuitBreaker.addOrUpdateRule(...rules: Rule[]): void

/**
 * 移除规则。
 */
kreex.circuitBreaker.removeRule(...rules: Rule[]): void

/**
 * 清除所有规则，breaker 恢复默认阈值，状态回到 closed。
 */
kreex.circuitBreaker.clearRules(): void

3. 强制操作

/**
 * 强制打开指定范围的 breaker。
 * scope 至少包含一个维度（nodeId/service/method/nodeName）。
 */
kreex.circuitBreaker.forceOpen(scope): void

/**
 * 强制关闭：恢复放行。
 */
kreex.circuitBreaker.forceClose(scope): void

/**
 * 强制重置：metrics 归零，状态回到 closed。
 */
kreex.circuitBreaker.forceReset(scope): void

4. 状态观察

/**
 * 查询 breaker 状态。breaker 未创建时返回 undefined。
 */
kreex.circuitBreaker.getState(key): CircuitState | undefined

/**
 * 查询实时指标：{ callCount, errorCount, errorRate, slowCallCount, slowCallRate }
 */
kreex.circuitBreaker.getMetrics(key): Metrics | undefined

/**
 * 查询当前阈值。
 */
kreex.circuitBreaker.getThresholds(key): BreakerThresholds | undefined

5. 桶策略

/**
 * 全局设置桶策略。
 */
kreex.circuitBreaker.breakerBarrel(policy?: BarrelPolicy): BarrelPolicy

// 存根级设置
serviceStub.breakerBarrel(policy): this

6. Rule 构造

import Kree4JS from '@kree4js/kree4js'
const { Rule } = Kree4JS.CircuitBreaker

/**
 * 构造规则：路由键 + 阈值。
 *
 * @param {CircuitBreakerKeyLike} key - 路由键：{ nodeId?, service?, method? }
 * @param {Partial<BreakerThresholds>} thresholds - 阈值（未设置的使用默认值）。
 */
new Rule(key, thresholds): Rule

7. Manager 事件

const { CircuitBreakerManagerEvent } = Kree4JS.CircuitBreaker

kreex.circuitBreaker.manager.on(CircuitBreakerManagerEvent.Opened, (payload) => {
  // payload: { nodeId, service, method, reason }
})

kreex.circuitBreaker.manager.on(CircuitBreakerManagerEvent.Closed, (payload) => { ... })
kreex.circuitBreaker.manager.on(CircuitBreakerManagerEvent.HalfOpened, (payload) => { ... })
kreex.circuitBreaker.manager.on(CircuitBreakerManagerEvent.StateChanged, (payload) => {
  // payload: { state, from, reason }
})

五. 可运行代码

完整示例代码，参见：14-circuit-breaker.mjs