聊聊resilience4j的CircuitBreakerConfig

序

本文主要研究一下resilience4j的CircuitBreakerConfig

CircuitBreakerConfig

resilience4j-circuitbreaker-0.13.0-sources.jar!/io/github/resilience4j/circuitbreaker/CircuitBreakerConfig.java

/**
 * A {@link CircuitBreakerConfig} configures a {@link CircuitBreaker}
 */
public class CircuitBreakerConfig {

    public static final int DEFAULT_MAX_FAILURE_THRESHOLD = 50; // Percentage
    public static final int DEFAULT_WAIT_DURATION_IN_OPEN_STATE = 60; // Seconds
    public static final int DEFAULT_RING_BUFFER_SIZE_IN_HALF_OPEN_STATE = 10;
    public static final int DEFAULT_RING_BUFFER_SIZE_IN_CLOSED_STATE = 100;
    public static final Predicate<Throwable> DEFAULT_RECORD_FAILURE_PREDICATE = (throwable) -> true;

    private float failureRateThreshold = DEFAULT_MAX_FAILURE_THRESHOLD;
    private int ringBufferSizeInHalfOpenState = DEFAULT_RING_BUFFER_SIZE_IN_HALF_OPEN_STATE;
    private int ringBufferSizeInClosedState = DEFAULT_RING_BUFFER_SIZE_IN_CLOSED_STATE;
    private Duration waitDurationInOpenState = Duration.ofSeconds(DEFAULT_WAIT_DURATION_IN_OPEN_STATE);
    // The default exception predicate counts all exceptions as failures.
    private Predicate<Throwable> recordFailurePredicate = DEFAULT_RECORD_FAILURE_PREDICATE;
    private boolean automaticTransitionFromOpenToHalfOpenEnabled = false;

    //......
}

這裏涉及了幾個參數以下：

• failureRateThreshold，默認爲50，即失敗率閾值爲50%
• ringBufferSizeInHalfOpenState，設置當斷路器處於HALF_OPEN狀態下的ring buffer的大小，它存儲了最近一段時間請求的成功失敗狀態，默認爲10
• ringBufferSizeInClosedState，設置當斷路器處於CLOSED狀態下的ring buffer的大小，它存儲了最近一段時間請求的成功失敗狀態，默認爲100
• waitDurationInOpenState，用來指定斷路器從OPEN到HALF_OPEN狀態等待的時長，默認是60秒
• recordFailurePredicate，用於判斷哪些異常應該算做失敗歸入斷路器統計，默認是Throwable類型
• automaticTransitionFromOpenToHalfOpenEnabled，當waitDurationInOpenState時間一過，是否自動從OPEN切換到HALF_OPEN，默認爲true

CircuitBreakerMetrics

resilience4j-circuitbreaker-0.13.0-sources.jar!/io/github/resilience4j/circuitbreaker/internal/CircuitBreakerMetrics.java

class CircuitBreakerMetrics implements CircuitBreaker.Metrics {

    private final int ringBufferSize;
    private final RingBitSet ringBitSet;
    private final LongAdder numberOfNotPermittedCalls;

    CircuitBreakerMetrics(int ringBufferSize) {
        this(ringBufferSize, null);
    }

    CircuitBreakerMetrics(int ringBufferSize, RingBitSet sourceSet) {
        this.ringBufferSize = ringBufferSize;
        if(sourceSet != null) {
            this.ringBitSet = new RingBitSet(this.ringBufferSize, sourceSet);
        }else{
            this.ringBitSet = new RingBitSet(this.ringBufferSize);
        }
        this.numberOfNotPermittedCalls = new LongAdder();
    }

    /**
     * Creates a new CircuitBreakerMetrics instance and copies the content of the current RingBitSet
     * into the new RingBitSet.
     *
     * @param targetRingBufferSize the ringBufferSize of the new CircuitBreakerMetrics instances
     * @return a CircuitBreakerMetrics
     */
    public CircuitBreakerMetrics copy(int targetRingBufferSize) {
        return new CircuitBreakerMetrics(targetRingBufferSize, this.ringBitSet);
    }

    /**
     * Records a failed call and returns the current failure rate in percentage.
     *
     * @return the current failure rate  in percentage.
     */
    float onError() {
        int currentNumberOfFailedCalls = ringBitSet.setNextBit(true);
        return getFailureRate(currentNumberOfFailedCalls);
    }

    /**
     * Records a successful call and returns the current failure rate in percentage.
     *
     * @return the current failure rate in percentage.
     */
    float onSuccess() {
        int currentNumberOfFailedCalls = ringBitSet.setNextBit(false);
        return getFailureRate(currentNumberOfFailedCalls);
    }

    /**
     * Records a call which was not permitted, because the CircuitBreaker state is OPEN.
     */
    void onCallNotPermitted() {
        numberOfNotPermittedCalls.increment();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public float getFailureRate() {
        return getFailureRate(getNumberOfFailedCalls());
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public int getMaxNumberOfBufferedCalls() {
        return ringBufferSize;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public int getNumberOfSuccessfulCalls() {
        return getNumberOfBufferedCalls() - getNumberOfFailedCalls();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public int getNumberOfBufferedCalls() {
        return this.ringBitSet.length();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public long getNumberOfNotPermittedCalls() {
        return this.numberOfNotPermittedCalls.sum();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public int getNumberOfFailedCalls() {
        return this.ringBitSet.cardinality();
    }

    private float getFailureRate(int numberOfFailedCalls) {
        if (getNumberOfBufferedCalls() < ringBufferSize) {
            return -1.0f;
        }
        return numberOfFailedCalls * 100.0f / ringBufferSize;
    }
}

• ringBitSet.setNextBit方法返回的是這個ringBitSet中被設置爲true的個數，這裏就是numberOfFailedCalls
• getFailureRate裏頭先判斷，若是ringBitSet中的個數沒有達到ringBufferSize指定的大小，則返回-1.0f，則不計算失敗率
• 失敗率的計算就是numberOfFailedCalls * 100.0f / ringBufferSize

RingBitSet

resilience4j-circuitbreaker-0.13.0-sources.jar!/io/github/resilience4j/circuitbreaker/internal/RingBitSet.java

/**
 * A ring bit set which stores bits up to a maximum size of bits.
 */
class RingBitSet {

    private final int size;
    private final BitSetMod bitSet;

    private boolean notFull;
    private int index = -1;

    private volatile int length;
    private volatile int cardinality = 0;

    //......

    /**
     * Sets the bit at the next index to the specified value.
     *
     * @param value a boolean value to set
     * @return the number of bits set to {@code true}
     */
    public synchronized int setNextBit(boolean value) {
        increaseLength();
        index = (index + 1) % size;

        int previous = bitSet.set(index, value);
        int current = value ? 1 : 0;
        cardinality = cardinality - previous + current;
        return cardinality;
    }

    private void increaseLength() {
        if (notFull) {
            int nextLength = length + 1;
            if (nextLength < size) {
                length = nextLength;
            } else {
                length = size;
                notFull = false;
            }
        }
    }
}

• 重點看setNextBit方法，這裏的cardinality就是ringbuffer中設置爲true的個數
• increaseLength方法，若是length不超過size則遞增，超過則重置爲size大小，即最大爲size大小，而後用notFull標記是否滿

HalfOpenState

resilience4j-circuitbreaker-0.13.0-sources.jar!/io/github/resilience4j/circuitbreaker/internal/HalfOpenState.java

final class HalfOpenState extends CircuitBreakerState {

    private CircuitBreakerMetrics circuitBreakerMetrics;
    private final float failureRateThreshold;

    HalfOpenState(CircuitBreakerStateMachine stateMachine) {
        super(stateMachine);
        CircuitBreakerConfig circuitBreakerConfig = stateMachine.getCircuitBreakerConfig();
        this.circuitBreakerMetrics = new CircuitBreakerMetrics(
                circuitBreakerConfig.getRingBufferSizeInHalfOpenState());
        this.failureRateThreshold = stateMachine.getCircuitBreakerConfig().getFailureRateThreshold();
    }

    /**
     * Returns always true, because the CircuitBreaker is half open.
     *
     * @return always true, because the CircuitBreaker is half open.
     */
    @Override
    boolean isCallPermitted() {
        return true;
    }

    @Override
    void onError(Throwable throwable) {
        // CircuitBreakerMetrics is thread-safe
        checkFailureRate(circuitBreakerMetrics.onError());
    }

    @Override
    void onSuccess() {
        // CircuitBreakerMetrics is thread-safe
        checkFailureRate(circuitBreakerMetrics.onSuccess());
    }

    /**
     * Checks if the current failure rate is above or below the threshold.
     * If the failure rate is above the threshold, transition the state machine to OPEN state.
     * If the failure rate is below the threshold, transition the state machine to CLOSED state.
     *
     * @param currentFailureRate the current failure rate
     */
    private void checkFailureRate(float currentFailureRate) {
        if(currentFailureRate != -1){
            if(currentFailureRate >= failureRateThreshold) {
                stateMachine.transitionToOpenState();
            }else{
                stateMachine.transitionToClosedState();
            }
        }
    }

    /**
     * Get the state of the CircuitBreaker
     */
    @Override
    CircuitBreaker.State getState() {
        return CircuitBreaker.State.HALF_OPEN;
    }

    @Override
    CircuitBreakerMetrics getMetrics() {
        return circuitBreakerMetrics;
    }
}

• 這裏的ringBufferSize爲getRingBufferSizeInHalfOpenState
• 這裏onError以及onSuccess都會觸發checkFailureRate，檢查失敗率是否達到閾值
• checkFailureRate判斷的時候首先排除掉-1.0f，而後若是達到閾值，則將狀態切換到OPEN，不達到閾值則切換到CLOSED狀態

ClosedState

resilience4j-circuitbreaker-0.13.0-sources.jar!/io/github/resilience4j/circuitbreaker/internal/ClosedState.java

final class ClosedState extends CircuitBreakerState {

    private final CircuitBreakerMetrics circuitBreakerMetrics;
    private final float failureRateThreshold;

    ClosedState(CircuitBreakerStateMachine stateMachine) {
        this(stateMachine, null);
    }

    ClosedState(CircuitBreakerStateMachine stateMachine, CircuitBreakerMetrics circuitBreakerMetrics) {
        super(stateMachine);
        CircuitBreakerConfig circuitBreakerConfig = stateMachine.getCircuitBreakerConfig();
        if(circuitBreakerMetrics == null){
            this.circuitBreakerMetrics = new CircuitBreakerMetrics(
                circuitBreakerConfig.getRingBufferSizeInClosedState());
        }else{
            this.circuitBreakerMetrics = circuitBreakerMetrics.copy(circuitBreakerConfig.getRingBufferSizeInClosedState());
        }
        this.failureRateThreshold = stateMachine.getCircuitBreakerConfig().getFailureRateThreshold();
    }

    /**
     * Returns always true, because the CircuitBreaker is closed.
     *
     * @return always true, because the CircuitBreaker is closed.
     */
    @Override
    boolean isCallPermitted() {
        return true;
    }

    @Override
    void onError(Throwable throwable) {
        // CircuitBreakerMetrics is thread-safe
        checkFailureRate(circuitBreakerMetrics.onError());
    }

    @Override
    void onSuccess() {
        // CircuitBreakerMetrics is thread-safe
        checkFailureRate(circuitBreakerMetrics.onSuccess());
    }

    /**
     * Checks if the current failure rate is above the threshold.
     * If the failure rate is above the threshold, transitions the state machine to OPEN state.
     *
     * @param currentFailureRate the current failure rate
     */
    private void checkFailureRate(float currentFailureRate) {
        if (currentFailureRate >= failureRateThreshold) {
            // Transition the state machine to OPEN state, because the failure rate is above the threshold
            stateMachine.transitionToOpenState();
        }
    }

    /**
     * Get the state of the CircuitBreaker
     */
    @Override
    CircuitBreaker.State getState() {
        return CircuitBreaker.State.CLOSED;
    }
    /**
     *
     * Get metrics of the CircuitBreaker
     */
    @Override
    CircuitBreakerMetrics getMetrics() {
        return circuitBreakerMetrics;
    }
}

• 這裏的ringBufferSize爲getRingBufferSizeInClosedState
• 這裏onError以及onSuccess都會觸發checkFailureRate，檢查失敗率是否達到閾值
• checkFailureRate判斷若是大於設定的閾值，則把狀態變爲OPEN

小結

• resilience4j的CircuitBreakerConfig主要設置了failureRateThreshold、ringBufferSizeInHalfOpenState、ringBufferSizeInClosedState、waitDurationInOpenState、recordFailurePredicate、automaticTransitionFromOpenToHalfOpenEnabled這幾個配置。
• ringBufferSizeInHalfOpenState、ringBufferSizeInClosedState設置的是在ringbuffer的大小，同時也影響failureRate的計算，即若是ringbuffer中的長度小於這個大小，則不會計算失敗率。
• ringbuffer中的length會遞增，可是不會超過指定的大小，最大爲指定的size

doc

• Resilience4j is a fault tolerance library designed for Java8 and functional programming