熔断器
熔断器模式
熔断器模式定义了熔断器开关相互转换的逻辑:
服务的健康状况 = 请求失败数 / 请求总数. 熔断器开关由关闭到打开的状态转换是通过当前服务健康状况和设定阈值比较决定的.
- 当熔断器开关关闭时, 请求被允许通过熔断器. 如果当前健康状况高于设定阈值, 开关继续保持关闭. 如果当前健康状况低于设定阈值, 开关则切换为打开状态.
- 当熔断器开关打开时, 请求被禁止通过.
- 当熔断器开关处于打开状态, 经过一段时间后, 熔断器会自动进入半开状态, 这时熔断器只允许一个请求通过. 当该请求调用成功时, 熔断器恢复到关闭状态. 若该请求失败, 熔断器继续保持打开状态, 接下来的请求被禁止通过.
熔断器的开关能保证服务调用者在调用异常服务时, 快速返回结果, 避免大量的同步等待. 并且熔断器能在一段时间后继续侦测请求执行结果, 提供恢复服务调用的可能.
资源上可以互相隔离。比如一个依赖服务端一个熔断器。
微服务可以两个熔断器，一个是针对服务实例熔断、一个是针对依赖服务接口熔断。
并且也不必全靠真实流量从半开到闭合，像实例也可以用Ping定时检测自动恢复，避免影响真实流量。
以上也是Hystrix [hɪst’rɪks]的设计理念。
熔断器实现
以golang为例子。使用无锁结构或者atomic方式。
维护bucket，比如每10ms一个bucket， bucket上维护计数器cnt。
每次接口调用都在最新的bucket上+1，可能是success也可能是fail。
计算是否熔断时累加每个bucket的值即可。bucket组织形式感觉可以用的循环数组实现。循环数组可以用个mutex锁住。
窗口时间 10~60 秒 统计滑动窗口时间
最小请求数 10~20 达到后才判断失败率
失败率阈值 50% 达到则触发熔断
熔断持续时间 30s~1min OPEN 状态维持时间
HALF-OPEN 请求 1~3 个 用于探测恢复
+---------+
| CLOSED |
+---------+
|
| 错误率 > 阈值
v
+---------+
| OPEN |
+---------+
|
| 超时恢复周期后
v
+-------------+
| HALF-OPEN |
+-------------+
| 请求成功 -> CLOSED
| 请求失败 -> OPEN
python实现
- 双端队列记录最近一分钟的成功次数和失败次数。读写锁保护更新。
- 当前状态， 如果是Open，然后距离上次失败时间超过了阈值，就开始半开。
- 上次失败的时间。
主要逻辑
- 只有close和半开才会请求访问。半开只会调用一次，因此半开后变成开或者关。
- 开时抛出异常。
优化，未实现
循环数组替代双端队列， 一分钟，就61个slot，其中一个一直为空，这样60个slot。 这样不需要读写锁。
每个元素都用atomic更新。
import time
import threading
from collections import deque
class TimeWindowCircuitBreaker:
class State:
CLOSED = 'closed'
OPEN = 'open'
HALF_OPEN = 'half_open'
def __init__(self, error_rate_threshold=0.5, window_seconds=60, min_calls=10, recovery_timeout=10, expected_exception=Exception):
self.error_rate_threshold = error_rate_threshold
self.window_seconds = window_seconds
self.min_calls = min_calls
self.recovery_timeout = recovery_timeout
self.expected_exception = expected_exception
self._state = self.State.CLOSED
self._last_failure_time = None
self._lock = threading.Lock()
self._events = deque() # [(timestamp, success: bool)]
def _prune_old_events(self):
now = time.time()
while self._events and self._events[0][0] < now - self.window_seconds:
self._events.popleft()
def _calculate_error_rate(self):
self._prune_old_events()
total = len(self._events)
if total == 0:
return 0.0, 0
failures = sum(1 for _, success in self._events if not success)
return failures / total, total
def call(self, func, *args, **kwargs):
now = time.time()
with self._lock:
# 如果是 OPEN 状态
if self._state == self.State.OPEN:
if now - self._last_failure_time > self.recovery_timeout:
self._state = self.State.HALF_OPEN
else:
raise Exception('CircuitBreaker is OPEN')
try:
result = func(*args, **kwargs)
except self.expected_exception as e:
with self._lock:
self._events.append((time.time(), False))
error_rate, total = self._calculate_error_rate()
if total >= self.min_calls and error_rate > self.error_rate_threshold:
self._state = self.State.OPEN
self._last_failure_time = time.time()
raise e
else:
with self._lock:
self._events.append((time.time(), True))
if self._state == self.State.HALF_OPEN:
self._state = self.State.CLOSED
return result
@property
def state(self):
with self._lock:
return self._state