공유 자원을 관리하는 bulk head에 대해서 알아보자

벌크헤드에 대해서 간단하게 알아보겠습니다

Bulk Head는 한 가지 작업에서 동시에 사용될 수 있는 자원을 제한하는 방식입니다.

서버 개발에서는 보통 한 API에서 최대 N개까지만 동시에 호출될 수 있다는 제약 조건을 걸게 됩니다.

톰캣 스레드(외부에서 요청을 받는 스레드)가 200개가 존재하고 있을 때, 한 API에서 총 199개까지만 호출될 수 있다고 하면, 1개는 무조건 다른 쪽에 배정될 테니 다른 기능이 최소한 어느 정도는 동작하는 것을 보장할 수 있습니다.

이 제한이 왜 필요한지 아직 전혀 감이 오지 않으시죠?

보통 다른 API 를 호출할 때, 몇 번 이상 호출하면 안 된다는 제약을 거는 것은 봤어도 우리 서비스에 동시에 몇 개 이상 호출이 되면 처리를 못하게 막는다는 것은 이상할 테니까요

이런 제한이 왜 필요한지에 대해서 지금부터 가상의 사례를 바탕으로 알아보도록 하겠습니다

상황 설명

저희 비즈니스는 2개의 인스턴스를 활용해서 작은 msa 환경을 구축해 총 2가지 기능을 제공하고 있습니다

첫 번째 서비스

기능 목록

1. 현재 티스토리에서 블로그 포스트를 총 몇 개나 썼는지 기록해 주는 api
   1. GET /api/tistorybank/posts라고 하는 가상의 api를 호출한다
      1. 호출을 했을 때, 3초 동안 응답이 없다면, 장애 판정을 내리고, 10초 동안 호출하지 않습니다
   2. 그 결과를 Json으로 전달해 줍니다
2. 가장 최근에 쓴 포스트 1개를 주기적으로 저장해 뒀다가, db에서 꺼내서 그대로 전달해 주는 서비스
   1. 이 결과도 Json으로 전달해 줍니다

두 번째 서비스

기능 목록

1. 첫 번째 서비스로부터 온 결과를 이용해 html을 만들어줍니다
   1. 총 포스트 수를 예쁘게 보여주는 기능이 있습니다
   2. 가장 최근에 쓴 포스트를 예쁘게 보여주는 기능이 있습니다.
2. 첫 번째 서비스의 api를 호출했을 때, 1초 이상 응답이 없으면 장애 판정을 내립니다.
   1. 장애 회복을 위해 10초 동안은 첫 번째 서비스를 호출하지 않습니다
3. 첫 번째 서비스가 장애가 났다면, 장애가 났다는 html을 만들어줍니다.

위와 같은 기능을 가진 2가지 인스턴스가 잘 작동하고 있는 상황을 생각해 봅시다

서비스를 운영하다가 갑자기 티스토리 api 가 장애가 발생했다고 생각해 봅시다

저희 비즈니스에서도 장애 알림이 오기 시작했습니다

첫 번째 서비스에서는 기능 1번(티스토리 블로그 글 개수 세주기)

두 번째 서비스에서는 첫 번째 서비스의 기능 1번을 호출할 수 없다는 장애 알람이 오기 시작했습니다

여기까지는 당연히 그럴 수 있죠?

하지만 여기서 예상하지 못했던 장애가 추가로 알림이 오기 시작합니다.

두 번째 서비스에서 첫 번째 서비스의 기능 2번도 작동을 하지 않는다는 알림이 도착했습니다.

어떤 과정을 통해서 저런 일들이 발생했는지 조금 더 자세하게 알아보도록 하겠습니다

1. 티스토리 API 가 장애가 발생해서, 작동을 멈춥니다
2. 첫 번째 서비스에서 3초간 응답을 기다리고 있습니다
3. 공유 자원이 모두 기능 1번을 기다리면서 고갈됩니다.
4. 첫 번째 서비스의 두 번째 기능도 비정상적으로 긴 대기시간이 생깁니다
5. 두 번째 서비스에서 첫 번째 서비스의 모든 기능이 작동하지 않는 것을 발견하고 모든 호출을 하지 않습니다

벌크헤드를 적용했을 때

모든 기능이 작동하지 않는 것이 아니라, 마지막 포스트를 보여주는 것은 아주 잘 동작합니다

최초의 설정

기본 톰캣 스레드 중 200개 모두 한 API에 사용될 수 있다.

그렇기 때문에 전체 기능이 동작하지 않는 문제가 발생하였습니다

두 번째 설정

기본 톰캣 스레드중 150개까지만 한 API에서 사용될 수 있다.

나머지 50개는 다른 API를 위해서 남겨두도록 설정했습니다.

그렇다면 이 설정으로 어떤 문제가 해결될 수 있는지 알아보도록 하겠습니다.

1. 티스토리 API 가 장애가 발생해서, 작동을 멈춥니다
2. 첫 번째 서비스에서 3초간 응답을 기다리고 있습니다
3. 공유 자원 중 150개의 톰캣 스레드가 모두 기능 1번을 기다리면서 고갈됩니다.
4. 첫 번째 서비스의 두 번째 기능은 나머지 50개의 톰캣 스레드를 활용해 처리가 됩니다
5. 두 번째 서비스에서 첫 번째 서비스의 기능 1번만 작동하지 않고, 2번은 정상 작동하는 것을 확인할 수 있습니다

벌크헤드를 사용하는 방법

스프링을 사용하신다면 아래와 같은 내용을 추가하시면 됩니다

implementation ("org.springframework.cloud:spring-cloud-starter-circuitbreaker-resilience4j")
//부트 3이라면
implementation ("io.github.resilience4j:resilience4j-spring-boot3")
///부트 2버전이라면
implementation ("io.github.resilience4j:resilience4j-bulkhead:${resilience4jVersion}")

AspectJ를 사용하고 계시지 않으시다면 AspectJ를 추가해주셔야 합니다

implementation ("org.aspectj:aspectjweaver:${aspectjVersion}")

벌크 헤드를 구현하는 방식은 정말 많지만 이번에는 resilience4 j를 위주로 다뤄보도록 하겠습니다

implementation ("io.github.resilience4j:resilience4j-bulkhead:${resilience4jVersion}")

버전 1.x 를 사용한다면 java 8로도 동작할 수 있게 설계되어 있습니다

https://github.com/resilience4j/resilience4j/issues/1598

2.x 는 아래에서 설명한 것처럼 java 17 버전이 필요합니다

가장 최신 버전은 2.1.0입니다

위와 같은 의존성을 추가하게 되면 벌크헤드를 사용할 수 있는데요

사용 방법은 정말 간단합니다

resilience4j.bulkhead:
  configs:
    default:
      maxConcurrentCalls: 1

위와 같은 yml 부분을 추가하면 최대 1개의 동시요청만이 허가됩니다.

@GetMapping("/bulkhead/1")    
@Bulkhead(name = "bulkhead")
public String bulkhead() {
    bulkheadService.call();
    return "bulkhead/1";
}

스프링에서 Resilience4j 의 구현

스프링은 클래스 path에 Resilience4j 가 있다면 자동으로 이를 래핑 한 클래스를 사용합니다.

이때 바뀌는 설정들이 몇 가지 있는데요

가장 중요하게 바뀌는 부분이 기본적으로 ThreadPool을 활용한 Bulkhead를 사용한다는 점입니다.

https://docs.spring.io/spring-cloud-circuitbreaker/docs/current/reference/html/#bulkhead-pattern-supporting

에서 확인할 수 있는 것처럼, 기본적으로 내부적으로 구성된 스레드풀을 사용하게 되는데, 이 부분은 따로 yml 옵션을 줘서 해결할 수 있습니다

라고 적혀있지만, 코드상에서는 기본 타입이 Semaphore로 되어있는 것으로 적혀있습니다

Resilience4j 에 있는 구현 방식

기본적으로 어노테이션에 있는 THREADPOOL 인지 아닌지를 보고, 그에 따라서 맞춰서 사용하게 되어있습니다

if (bulkheadAnnotation.type() == Bulkhead.Type.THREADPOOL) {
    final CheckedSupplier<Object> bulkheadExecution =
        () -> proceedInThreadPoolBulkhead(proceedingJoinPoint, methodName, returnType, backend);
    return fallbackExecutor.execute(proceedingJoinPoint, method, bulkheadAnnotation.fallbackMethod(), bulkheadExecution);
} else {
    io.github.resilience4j.bulkhead.Bulkhead bulkhead = getOrCreateBulkhead(methodName,
        backend);
    final CheckedSupplier<Object> bulkheadExecution = () -> proceed(proceedingJoinPoint, methodName, bulkhead, returnType);
    return fallbackExecutor.execute(proceedingJoinPoint, method, bulkheadAnnotation.fallbackMethod(), bulkheadExecution);
}

아래와 같이, 전용 스레드풀을 만들고, 그 스레드풀의 큐에 요청을 쌓아두다, 일정 사이즈가 넘어가면 예외를 발생시키는 구조가 되어있습니다.

당연하지만 매번 스레드풀을 만드는 오버헤드가 발생합니다.

ThreadPoolBulkhead threadPoolBulkhead = threadPoolBulkheadRegistry.bulkhead(backend);
if (CompletionStage.class.isAssignableFrom(returnType)) {
    // threadPoolBulkhead.executeSupplier throws a BulkheadFullException, if the Bulkhead is full.
    // The RuntimeException is converted into an exceptionally completed future
    try {
        return threadPoolBulkhead.executeCallable(() -> {
            try {
                return ((CompletionStage<?>) proceedingJoinPoint.proceed())
                    .toCompletableFuture().get();
            } catch (ExecutionException e) {
                throw new CompletionException(e.getCause());
            } catch (InterruptedException | CancellationException e) {
                throw e;
            } catch (Throwable e) {
                throw new CompletionException(e);
            }
        });
    } catch (BulkheadFullException ex){
        CompletableFuture<?> future = new CompletableFuture<>();
        future.completeExceptionally(ex);
        return future;
    }
} else {
    throw new IllegalStateException(
        "ThreadPool bulkhead is only applicable for completable futures ");
}

반대로 Semaphore 방식은 스레드풀을 사용하지 않기에, 오버헤드가 적습니다

Hyxtrix에서 Resilience4j 로 옮겨가는 이유 중 위와 같은 오버헤드도 큰 부분을 차지한다고 합니다

Semaphore 방식은 현재 스레드에서 코드를 실행하는데, 앞 뒤에 semaphore를 관리하는 방식으로 처리하게 됩니다.

Bulkhead를 직접 테스트해 보자

// https://mvnrepository.com/artifact/io.rest-assured/rest-assured
testImplementation("io.rest-assured:rest-assured:5.3.2")

가장 먼저 rest-assured를 추가합니다

@SpringBootTest(webEnvironment = WebEnvironment.RANDOM_PORT)
public abstract class AcceptanceBase {

    @LocalServerPort
    private int port;

    @BeforeEach
    void setRestAssured() {
        RestAssured.port = port;
    }
}

컨트롤러의 코드는 다음과 같습니다. 요청이 하나 들어오면 1초간 기다렸다가, 응답으로 문자열을 줍니다

@GetMapping("/bulkhead/1")
@Bulkhead(name = "bulkhead")
public String bulkhead() {
    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    }
    return "bulkhead/1";
}

yml 설정은 다음과 같습니다

resilience4j.bulkhead:
  configs:
    default:
      maxConcurrentCalls: 1
      max-wait-duration: 2000ms

요청이 동시에 들어갈 수 있는 최대치는 1개이고

2번째 요청이 기다릴 수 있는 최대 시간은 2초입니다.

바로 요청이 2개 다 들어가면 안 된다는 점을 테스트코드로 작성하면 다음과 같습니다

@SuppressWarnings("NonAsciiCharacters")
class BulkheadControllerTest extends AcceptanceBase {

    @Test
    void bulkhead_가_동시에_1번만_실행_시킬_수_있도록_만든다() throws InterruptedException {
        // given
        ExecutorService executorService = Executors.newFixedThreadPool(2);
        long startTime = System.currentTimeMillis();

        // when
        executorService.execute(bulkhead());
        executorService.execute(bulkhead());

        // then
        executorService.shutdown();
        executorService.awaitTermination(3, TimeUnit.SECONDS);
        assertThat(System.currentTimeMillis() - startTime).isGreaterThan(2000);
    }

    private Runnable bulkhead() {
        return () -> RestAssured.given().log().all()
            .when().get("/bulkhead/1")
            .then().log().all()
            .statusCode(200);
    }
}

2개의 요청이 병렬로 처리되는 것이 아닌, 순차적으로 처리되었기 때문에, 각각 1초씩 총 2초가 넘게 걸린다는 부분을 위주로 테스트 코드를 작성했습니다

 

긴 글을 읽어주셔서 감사합니다