CircuitBreaker 斷路器

服務(wù)熔斷是為了保護(hù)我們的服務(wù)，比如當(dāng)某個(gè)服務(wù)出現(xiàn)問(wèn)題的時(shí)候，控制打向它的流量，讓它有時(shí)間去恢復(fù)，或者限制一段時(shí)間只能有固定數(shù)量的請(qǐng)求打向這個(gè)服務(wù)。這些都是保護(hù)措施。我在實(shí)際工作中也確實(shí)遇到過(guò)，數(shù)據(jù)庫(kù)出現(xiàn)問(wèn)題了，進(jìn)而導(dǎo)致Web服務(wù)出現(xiàn)問(wèn)題了，導(dǎo)致不依賴數(shù)據(jù)庫(kù)的服務(wù)也出現(xiàn)問(wèn)題了，出現(xiàn)一連串問(wèn)題。 這次學(xué)習(xí)《玩轉(zhuǎn) Spring 全家桶》，丁雪豐老師給了使用resilience4j的例子。 丁老師的例子是2019年的，這個(gè)框架已經(jīng)修改了些方法，所以我自己也花了些時(shí)間來(lái)理解了它的用法?，F(xiàn)將過(guò)程記錄下來(lái)。
首先POM文件引入

<dependency>
    <groupId>io.github.resilience4j</groupId>
    <artifactId>resilience4j-spring-boot2</artifactId>
    <version>2.0.2</version>
</dependency>

接著改造之前的Controller方法

@RestController
@RequestMapping("/customer")
@Slf4j
public class BookController {

    @Autowired
    private BookService bookService;

    private CircuitBreaker circuitBreaker;

    public BookController(CircuitBreakerRegistry registry) {
        circuitBreaker = registry.circuitBreaker("menu");
    }

    @GetMapping("/menu")
    public List<Book> readMenu() {
        Supplier<List<Book>> supplier = () -> bookService.getAll();
        circuitBreaker.getEventPublisher()
                .onEvent(event -> log.info(event.toString()));
        try{
            return circuitBreaker.executeSupplier(supplier);
        }
        catch (Exception ex)
        {
            log.error(ex.getMessage());
            return Collections.emptyList();
        }
    }
}

不同的地方就是引入了CircuitBreaker， 然后使用它將我們的方法“bookService.getAll()”包起來(lái)了。
然后在配置文件中添加如下的配置

resilience4j.circuitbreaker.backends.menu.failure-rate-threshold=50
resilience4j.circuitbreaker.backends.menu.wait-duration-in-open-state=60000
resilience4j.circuitbreaker.backends.menu.sliding-window-size=5
resilience4j.circuitbreaker.backends.menu.permitted-number-of-calls-in-half-open-state=2
resilience4j.circuitbreaker.backends.menu.minimum-number-of-calls=2

稍微解釋一下這里的配置
failure-rate-threshold=50是說(shuō)失敗率超過(guò)50%就熔斷，
wait-duration-in-open-state= 60000，是說(shuō)熔斷后等待60S才允許再次調(diào)用。
sliding-window-size =5 可以理解為5個(gè)請(qǐng)求統(tǒng)計(jì)一次，
permitted-number-of-calls-in-half-open-state = 2是說(shuō)進(jìn)入半開的狀態(tài)的時(shí)候，還允許請(qǐng)求多少個(gè)。
minimum-number-of-calls=2是說(shuō)最少有多少個(gè)請(qǐng)求才開始統(tǒng)計(jì)。 這里的參數(shù)都是我為了實(shí)驗(yàn)設(shè)置的，實(shí)際情況根據(jù)需要進(jìn)行調(diào)整。參數(shù)比較多，具體可以參加官方文檔
https://resilience4j.readme.io/docs/circuitbreaker

我們來(lái)看下實(shí)際的效果通過(guò)瀏覽器訪問(wèn)，
首先我們現(xiàn)打開BookService，讓它有一次成功的請(qǐng)求，日志會(huì)輸出
CircuitBreaker 'menu' recorded a successful call.
然后我們將BookService關(guān)閉，讓它請(qǐng)求失敗，日志會(huì)輸出如下
CircuitBreaker 'menu' recorded an error: 'feign.RetryableException: Connection refused: no further information executing GET http://bookshop-service/book/getAll'. Elapsed time: 2050 ms
CircuitBreaker 'menu' exceeded failure rate threshold. Current failure rate: 50.0
CircuitBreaker 'menu' changed state from CLOSED to OPEN
可以看到斷路器已經(jīng)打開了，
接著我們繼續(xù)訪問(wèn)會(huì)出現(xiàn)，
CircuitBreaker 'menu' recorded a call which was not permitted.
這個(gè)時(shí)候請(qǐng)求不會(huì)打到BookService上面了。就算這個(gè)時(shí)候我們的BookService恢復(fù)正常。
等待60s后進(jìn)入半Open的狀態(tài)
CircuitBreaker 'menu' changed state from OPEN to HALF_OPEN
這個(gè)時(shí)候恢復(fù)BookService正常，我們請(qǐng)求也會(huì)正常響應(yīng)了
CircuitBreaker 'menu' recorded a successful call
多請(qǐng)求幾次，斷路器就從HALF_OPEN變成了CLOSED
CircuitBreaker 'menu' changed state from HALF_OPEN to CLOSED

這里給一個(gè)官方的狀態(tài)圖來(lái)說(shuō)明

斷路器有三個(gè)狀態(tài)： CLOSED， OPEN， HALF_OPEN。

CLOSED是最開始的狀態(tài)，也就是關(guān)閉狀態(tài)，流量可以正常通過(guò)，當(dāng)失敗比率超過(guò)threshold后，斷路器打開， 變成OPEN 打開后流量不可以通過(guò)；等待一定的時(shí)間后，斷路器進(jìn)入半開狀態(tài) HALF_OPEN, 這個(gè)時(shí)候如果失敗率低于閾值，斷路器進(jìn)入CLOSED狀態(tài)，如果超過(guò)閾值，斷路器繼續(xù)保證OPEN，再等待，如此往復(fù)。

斷路器現(xiàn)在還支持設(shè)置慢請(qǐng)求，使用起來(lái)還是比較方便。對(duì)于參數(shù)的設(shè)置如果不是很理解，可以通過(guò)單元測(cè)試的方法來(lái)加深對(duì)它的理解。這里參考https://github.com/eugenp/tutorials/blob/master/libraries-6/src/test/java/com/baeldung/resilence4j/Resilience4jUnitTest.java 上面的例子，給出來(lái)個(gè)單元測(cè)試

    interface RemoteService {

        int process(int i);
    }

    private RemoteService service;

    @Test
    public void whenCircuitBreakerIsUsed_thenItWorksAsExpected() {
        service = mock(RemoteService.class);
        CircuitBreakerConfig config = CircuitBreakerConfig.custom()
                // Percentage of failures to start short-circuit
                .failureRateThreshold(20)
                .minimumNumberOfCalls(5)
                .build();
        CircuitBreakerRegistry registry = CircuitBreakerRegistry.of(config);
        CircuitBreaker circuitBreaker = registry.circuitBreaker("my");
        Function<Integer, Integer> decorated = CircuitBreaker.decorateFunction(circuitBreaker, service::process);

        when(service.process(anyInt())).thenThrow(new RuntimeException());
        circuitBreaker.getEventPublisher()
                .onEvent(event ->
                {
                    log.info(event.toString());
                });

        for (int i = 0; i < 10; i++) {
            try {
                decorated.apply(i);
            } catch (Exception ignore) {
            }
        }

        verify(service, times(5)).process(any(Integer.class));
    }

這里設(shè)置最少請(qǐng)求5次，失敗率超過(guò)20%就熔斷，然后我們請(qǐng)求了10次，實(shí)際上只調(diào)用了Service5次。
對(duì)于其它參數(shù)，你可以調(diào)整后，根據(jù)需要來(lái)驗(yàn)證是否符合預(yù)期。它的日志輸出如下

CircuitBreaker 'my' recorded an error: 'java.lang.RuntimeException'. Elapsed time: 2 ms
CircuitBreaker 'my' recorded an error: 'java.lang.RuntimeException'. Elapsed time: 0 ms
CircuitBreaker 'my' recorded an error: 'java.lang.RuntimeException'. Elapsed time: 0 ms
CircuitBreaker 'my' recorded an error: 'java.lang.RuntimeException'. Elapsed time: 0 ms
CircuitBreaker 'my' recorded an error: 'java.lang.RuntimeException'. Elapsed time: 0 ms
CircuitBreaker 'my' exceeded failure rate threshold. Current failure rate: 100.0
CircuitBreaker 'my' changed state from CLOSED to OPEN
CircuitBreaker 'my' recorded a call which was not permitted.
CircuitBreaker 'my' recorded a call which was not permitted.
CircuitBreaker 'my' recorded a call which was not permitted.
CircuitBreaker 'my' recorded a call which was not permitted.
CircuitBreaker 'my' recorded a call which was not permitted.  

可以看到5次過(guò)后，就開始打開斷路器，后面的call就不被允許了。

隔艙Bulkhead

Resilience4j 里面的Bulkhead可以簡(jiǎn)單的理解為允許多少個(gè)并發(fā)訪問(wèn)。我們這里還是通過(guò)單元測(cè)試的方法來(lái)演示它的功能

    @Test
    public void whenBulkheadIsUsed_thenItWorksAsExpected() throws InterruptedException {
        service = mock(RemoteService.class);
        BulkheadConfig config = BulkheadConfig.custom().maxConcurrentCalls(2).build();
        BulkheadRegistry registry = BulkheadRegistry.of(config);
        Bulkhead bulkhead = registry.bulkhead("my");
        Function<Integer, Integer> decorated = Bulkhead.decorateFunction(bulkhead, service::process);

       try {
            callAndBlock(decorated);
        }
       catch(BulkheadFullException ex)
        {
            log.error("isfull");
        }
        finally
        {
           verify(service, times(2)).process(any(Integer.class));
        }

    }

    private void callAndBlock(Function<Integer, Integer> decoratedService) throws InterruptedException {
        when(service.process(anyInt())).thenAnswer(invocation -> {
            log.info("service called");
            return null;
        });

        ArrayList<Integer> numberList = new ArrayList<Integer>();
        for(int i = 0;i<10;i++)
        {
            numberList.add(i);
        }

        numberList.parallelStream().forEach((i)->{
            try {
                decoratedService.apply(i);
            }
            catch (Exception ex)
            {
                log.error("meet error " + ex.getMessage());
            } catch (Throwable e) {
                throw new RuntimeException(e);
            }
        });
    }

首先我們解讀一下callAndBlock， 它會(huì)并發(fā)的去執(zhí)行一個(gè)function. 如果我們不用隔艙，它的輸出會(huì)是這樣。

2022-12-28T15:22:52.010+08:00  INFO 37276 --- [onPool-worker-4] c.k.r.bookcustomer.Resilience4jUnitTest  : service called
2022-12-28T15:22:52.010+08:00  INFO 37276 --- [onPool-worker-9] c.k.r.bookcustomer.Resilience4jUnitTest  : service called
2022-12-28T15:22:52.010+08:00  INFO 37276 --- [onPool-worker-5] c.k.r.bookcustomer.Resilience4jUnitTest  : service called
2022-12-28T15:22:52.010+08:00  INFO 37276 --- [onPool-worker-3] c.k.r.bookcustomer.Resilience4jUnitTest  : service called
2022-12-28T15:22:52.010+08:00  INFO 37276 --- [onPool-worker-1] c.k.r.bookcustomer.Resilience4jUnitTest  : service called
2022-12-28T15:22:52.010+08:00  INFO 37276 --- [onPool-worker-6] c.k.r.bookcustomer.Resilience4jUnitTest  : service called
2022-12-28T15:22:52.010+08:00  INFO 37276 --- [onPool-worker-7] c.k.r.bookcustomer.Resilience4jUnitTest  : service called
2022-12-28T15:22:52.010+08:00  INFO 37276 --- [onPool-worker-8] c.k.r.bookcustomer.Resilience4jUnitTest  : service called
2022-12-28T15:22:52.010+08:00  INFO 37276 --- [onPool-worker-2] c.k.r.bookcustomer.Resilience4jUnitTest  : service called
2022-12-28T15:22:52.011+08:00  INFO 37276 --- [           main] c.k.r.bookcustomer.Resilience4jUnitTest  : service called

可以看到啟動(dòng)了10個(gè)線程去訪問(wèn)方法。加了隔艙后，隔艙限定了一次只能兩個(gè)，輸出如下

2022-12-28T15:33:48.648+08:00 ERROR 32256 --- [onPool-worker-4] c.k.r.bookcustomer.Resilience4jUnitTest  : meet error Bulkhead 'my' is full and does not permit further calls
2022-12-28T15:33:48.648+08:00 ERROR 32256 --- [onPool-worker-6] c.k.r.bookcustomer.Resilience4jUnitTest  : meet error Bulkhead 'my' is full and does not permit further calls
2022-12-28T15:33:48.648+08:00 ERROR 32256 --- [onPool-worker-7] c.k.r.bookcustomer.Resilience4jUnitTest  : meet error Bulkhead 'my' is full and does not permit further calls
2022-12-28T15:33:48.648+08:00 ERROR 32256 --- [onPool-worker-4] c.k.r.bookcustomer.Resilience4jUnitTest  : meet error Bulkhead 'my' is full and does not permit further calls
2022-12-28T15:33:48.648+08:00 ERROR 32256 --- [onPool-worker-8] c.k.r.bookcustomer.Resilience4jUnitTest  : meet error Bulkhead 'my' is full and does not permit further calls
2022-12-28T15:33:48.648+08:00 ERROR 32256 --- [onPool-worker-5] c.k.r.bookcustomer.Resilience4jUnitTest  : meet error Bulkhead 'my' is full and does not permit further calls
2022-12-28T15:33:48.648+08:00 ERROR 32256 --- [onPool-worker-2] c.k.r.bookcustomer.Resilience4jUnitTest  : meet error Bulkhead 'my' is full and does not permit further calls
2022-12-28T15:33:48.648+08:00 ERROR 32256 --- [onPool-worker-3] c.k.r.bookcustomer.Resilience4jUnitTest  : meet error Bulkhead 'my' is full and does not permit further calls
2022-12-28T15:33:48.650+08:00  INFO 32256 --- [onPool-worker-1] c.k.r.bookcustomer.Resilience4jUnitTest  : service called
2022-12-28T15:33:48.650+08:00  INFO 32256 --- [           main] c.k.r.bookcustomer.Resilience4jUnitTest  : service called

可以看到只有兩次成功的訪問(wèn)，其它的訪問(wèn)都被block了。

限速器RateLimiter

RateLimiter的功能是限定一段時(shí)間內(nèi)允許多少次訪問(wèn)，還是使用和Bulkhead一樣的例子一樣

 @Test
    public void whenRateLimiterInUse_thenItWorksAsExpected() throws InterruptedException {
        service = mock(RemoteService.class);

        RateLimiterConfig config = RateLimiterConfig.custom()
                .limitRefreshPeriod(Duration.ofMillis(1000))
                .limitForPeriod(4)
                .timeoutDuration(Duration.ofMillis(25))
                .build();

        RateLimiterRegistry rateLimiterRegistry = RateLimiterRegistry.of(config);

        RateLimiter rateLimiter = rateLimiterRegistry
                .rateLimiter("name1");

        CheckedFunction<Integer, Integer> decorated = RateLimiter
                .decorateCheckedFunction(rateLimiter, service::process);

        try {
            callAndBlock(decorated);
        }
        catch(Exception ex)
        {
            log.error("isfull");
        }
        finally
        {
            verify(service, times(4)).process(any(Integer.class));
        }

    }

    private void callAndBlock(CheckedFunction<Integer, Integer> decoratedService) throws InterruptedException {
        when(service.process(anyInt())).thenAnswer(invocation -> {
            log.info("service called");
            return null;
        });

        ArrayList<Integer> numberList = new ArrayList<Integer>();
        for(int i = 0;i<10;i++)
        {
            numberList.add(i);
        }

        numberList.parallelStream().forEach((i)->{
            try {
                decoratedService.apply(i);
            }
            catch (Exception ex)
            {
                log.error("meet error " + ex.getMessage());
            } catch (Throwable e) {
                throw new RuntimeException(e);
            }
        });
    }

我們這里故意設(shè)置1S中允許訪問(wèn)4次，實(shí)際的運(yùn)行情況也是只允許了4次。日志輸出如下

2022-12-28T15:39:52.027+08:00  INFO 35236 --- [onPool-worker-2] c.k.r.bookcustomer.Resilience4jUnitTest  : service called
2022-12-28T15:39:52.027+08:00  INFO 35236 --- [onPool-worker-5] c.k.r.bookcustomer.Resilience4jUnitTest  : service called
2022-12-28T15:39:52.027+08:00  INFO 35236 --- [           main] c.k.r.bookcustomer.Resilience4jUnitTest  : service called
2022-12-28T15:39:52.027+08:00  INFO 35236 --- [onPool-worker-7] c.k.r.bookcustomer.Resilience4jUnitTest  : service called
2022-12-28T15:39:52.053+08:00 ERROR 35236 --- [onPool-worker-6] c.k.r.bookcustomer.Resilience4jUnitTest  : meet error RateLimiter 'name1' does not permit further calls
2022-12-28T15:39:52.060+08:00 ERROR 35236 --- [onPool-worker-3] c.k.r.bookcustomer.Resilience4jUnitTest  : meet error RateLimiter 'name1' does not permit further calls
2022-12-28T15:39:52.060+08:00 ERROR 35236 --- [onPool-worker-9] c.k.r.bookcustomer.Resilience4jUnitTest  : meet error RateLimiter 'name1' does not permit further calls
2022-12-28T15:39:52.060+08:00 ERROR 35236 --- [onPool-worker-1] c.k.r.bookcustomer.Resilience4jUnitTest  : meet error RateLimiter 'name1' does not permit further calls
2022-12-28T15:39:52.060+08:00 ERROR 35236 --- [onPool-worker-8] c.k.r.bookcustomer.Resilience4jUnitTest  : meet error RateLimiter 'name1' does not permit further calls
2022-12-28T15:39:52.075+08:00 ERROR 35236 --- [onPool-worker-4] c.k.r.bookcustomer.Resilience4jUnitTest  : meet error RateLimiter 'name1' does not permit further calls

限速器這個(gè)功能只能限制在整體性能上面，如果要限制某個(gè)用戶，只能某段時(shí)間訪問(wèn)多少次，它就做不到了。

Relilience4j 里面還提供了Retry,TimeLimiter,Cache. 感覺(jué)不是很有必要的功能， Retry在spring里面有相應(yīng)的功能了，沒(méi)有必要專門為了使用它而多加個(gè)包。 TimeLimiter，Cache 我感覺(jué)不是很受重視的功能，連例子文檔都懶得提供，可見意義不大。

到此這篇關(guān)于Spring Cloud 使用 Resilience4j 實(shí)現(xiàn)服務(wù)熔斷的方法的文章就介紹到這了,更多相關(guān)Spring Cloud 服務(wù)熔斷內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論