Bulkhead — Tek Yavaş Servis Neden Herkesi Düşürür?

1
resilience4j:
2
  thread-pool-bulkhead:
3
    instances:
4
      cardProvider:
5
        core-thread-pool-size: 4
6
        max-thread-pool-size: 4
7
        queue-capacity: 0          # no hidden waiting room: a full pool answers at once
8
      balance:
9
        core-thread-pool-size: 3
10
        max-thread-pool-size: 3
11
        queue-capacity: 0
12
  timelimiter:
13
    instances:
14
      cardProvider:
15
        timeout-duration: 2s

1
@Component
2
class CardProviderClient {
3

4
    private final RestClient cards;
5

6
    CardProviderClient(RestClient.Builder builder) {
7
        this.cards = builder.baseUrl("https://card-provider.internal").build();
8
    }
9

10
    // Runs in its own pool of four threads and gives up after two seconds.
11
    @Bulkhead(name = "cardProvider", type = Bulkhead.Type.THREADPOOL, fallbackMethod = "busy")
12
    @TimeLimiter(name = "cardProvider")
13
    CompletableFuture<CardLimits> limits(String cardId) {
14
        return CompletableFuture.completedFuture(
15
                cards.get().uri("/cards/{id}/limits", cardId).retrieve().body(CardLimits.class));
16
    }
17

18
    private CompletableFuture<CardLimits> busy(String cardId, Throwable cause) {
19
        return CompletableFuture.completedFuture(CardLimits.temporarilyUnavailable(cardId));
20
    }
21
}

1
@Component
2
class BalanceClient {
3

4
    private final RestClient accounts;
5

6
    BalanceClient(RestClient.Builder builder) {
7
        this.accounts = builder.baseUrl("https://accounts.internal").build();
8
    }
9

10
    // A compartment of its own: a slow card provider cannot take these threads.
11
    @Bulkhead(name = "balance", type = Bulkhead.Type.THREADPOOL)
12
    CompletableFuture<Balance> balance(String iban) {
13
        return CompletableFuture.completedFuture(
14
                accounts.get().uri("/accounts/{iban}/balance", iban).retrieve().body(Balance.class));
15
    }
16
}

1
class BulkheadIsolationTest {
2

3
    @Test
4
    void aFullCardCompartmentLeavesBalanceThreadsFree() throws Exception {
5
        ThreadPoolBulkhead card = ThreadPoolBulkhead.of("card",
6
                ThreadPoolBulkheadConfig.custom().coreThreadPoolSize(1).maxThreadPoolSize(1).queueCapacity(0).build());
7
        ThreadPoolBulkhead balance = ThreadPoolBulkhead.of("balance",
8
                ThreadPoolBulkheadConfig.custom().coreThreadPoolSize(1).maxThreadPoolSize(1).queueCapacity(0).build());
9

10
        CountDownLatch release = new CountDownLatch(1);
11
        card.executeSupplier(() -> { awaitQuietly(release); return "slow"; });   // fills the card compartment
12

13
        assertThatThrownBy(() -> card.executeSupplier(() -> "second"))
14
                .isInstanceOf(BulkheadFullException.class);
15
        assertThat(balance.executeSupplier(() -> "ok").toCompletableFuture().get()).isEqualTo("ok");
16
        release.countDown();
17
    }
18

19
    private static void awaitQuietly(CountDownLatch latch) {
20
        try {
21
            latch.await();
22
        } catch (InterruptedException e) {
23
            Thread.currentThread().interrupt();
24
        }
25
    }
26
}