@KafkaListener의 concurrency 설정(feat.메시지 병렬 처리)

개요

@KafkaListener의 속성 중에는 concurrency 설정이 존재한다. 이 설정은 Kafka 리스너의 병렬 소비(Parallel Consumption)를 설정하는 옵션으로, 하나의 Kafka 토픽에 대해 여러 개의 소비자 스레드를 통해 병렬로 메시지를 처리할 수 있도록 하는 기능이다.

그러나 단순히 concurrency 값을 높인다고 해서 반드시 성능이 향상되는 것은 아니다. 특정 상황에서는 병렬 처리가 효과적이지만, 적절하지 않은 경우 오히려 불필요한 리소스 낭비로 이어질 수 있다. 따라서 이번 게시글에서는 concurrency의 개념과 예제를 통해 효과적으로 병렬 처리를 활용할 수 있는 경우를 살펴보고자 한다.

 

 

Concurrency 기본 개념

Kafka는 기본적으로 컨슈머 그룹(Consumer Group) 단위로 메시지를 할당한다. 그리고 동일한 컨슈머 그룹 내에서는 각 파티션이 하나의 컨슈머에게만 할당되며, 여러 컨슈머가 동일한 파티션을 중복으로 소비할 수 없다.

그림 1

위와 같이 Consumer-1과 Consumer-2가 존재하고, Topic-01와 Topic-02의 여러 파티션이 각각 이 두 컨슈머에게 할당되는 구조를 살펴보자. 기본 설정에서는 각 컨슈머가 할당받은 파티션에서 메시지를 하나씩 순차적으로 처리하는데, Consumer-1이 Topic-01의 Partition-1에서 메시지를 처리한 후 Topic-02의 Partition-1의 메시지를 처리하거나 그 반대 순서로 처리하는 방식이다. 이는 Kafka 컨슈머 스레드의 기본 설정(concurrency)이 1개이기 때문이다. (참고로 메시지를 처리하는 순서는 순차적이지만, poll()을 호출할 때는 여러 파티션에서 한꺼번에 메시지를 가져올 수 있다. 즉 가져오는 것은 동시에 가능하지만 처리 자체는 하나씩 이루어진다.)

그림 2

이 경우도 마찬가지다. Consumer-1이 Topic-01의 세 개의 파티션을 할당받으면, 하나의 스레드가 각 파티션들의 메시지를 순차적으로 처리한다.

그런데 잘 생각해 보자. 그림 1번에서 Topic-01와 Topic-02는 서로 독립적인 개념이므로 반드시 순차적으로 처리할 필요는 없을 것이다. 또한 그림 2에서도 메시지 처리 순서가 중요하지 않다면 여러 파티션을 동시에 처리하는 것이 더 효율적일 수 있을 것이다.

이를 위해 @KafkaListener에는 concurrency 속성을 제공하는데, 이 값을 설정하면 컨슈머 스레드 개수를 손쉽게 지정할 수 있다. concurrency 값을 지정하면 내부적으로 여러 개의 Kafka 컨슈머 스레드가 생성되며, 하나의 리스너가 여러 개의 파티션을 동시에 소비할 수 있어 병렬 처리가 가능해진다.

@KafkaListener(topics = "my-topic", concurrency = "2")
public void listen(String message) {
    System.out.println("Received: " + message);
}

위와 같이 concurrency 값을 2로 설정하면 하나의 리스너에서 두 개의 컨슈머 스레드가 실행되어 병렬로 메시지를 처리할 수 있게 된다.

 

 

concurrency 개수가 높으면 무조건 좋을까?

concurrency 값을 높이면 더 많은 Kafka 컨슈머 스레드가 생성되며, 여러 파티션을 메시지를 동시에 처리할 수 있어 처리 속도가 향상된다. 하지만 무조건 높은 값을 설정한다고 해서 항상 최적의 성능을 보장하는 것은 아니다. 앞서 설명한 개념에서 눈치챘을 수도 있겠지만, concurrency 값을 설정할 때는 반드시 할당받은 파티션의 개수를 고려해야 한다. Kafka의 메시지 소비는 파티션 단위로 이루어지기 때문에 concurrency 값이 파티션 개수를 초과하면 성능 향상에는 도움이 되지 않는다.

예시를 통해 직접 살펴보자.

1. 토픽이 1개일 경우(메시지 3개)

• 파티션 1개, concurrency 1
• 파티션 1개, concurrency 2
• 파티션 3개, concurrency 6

1. 토픽 1개(파티션 1개, concurrency 1)

Received message in group 'test': Test Message 0 | Processed by: Thread-1
Received message in group 'test': Test Message 1 | Processed by: Thread-1
Received message in group 'test': Test Message 2 | Processed by: Thread-1

Kafka 토픽에 하나의 파티션만 존재하는 상태에서 concurrency 값을 1로 설정한 경우, 단일 스레드가 모든 메시지를 순차적으로 처리하게 된다. 실제로 3개의 메시지를 발행한 후 로그를 확인해 보면, 하나의 스레드가 모든 메시지를 담당하고 있는 걸 확인할 수 있다.

2. 토픽 1개(파티션 1개, concurrency 2)

Received message in group 'test': Test Message 0 | Processed by: Thread-1
Received message in group 'test': Test Message 1 | Processed by: Thread-1
Received message in group 'test': Test Message 2 | Processed by: Thread-1

이제 concurrency 값을 2로 증가시켜 보았다. 직관적으로 생각하면 두 개의 스레드가 생성되어, 병렬 처리를 통해 소요 시간이 단축될 것처럼 보인다. 하지만 실제로는 여전히 하나의 스레드만 메시지를 소비한다. 이유는 동일한 파티션이기 때문에 하나의 스레드만 해당 파티션을 독점적으로 가져가고, 나머지 스레드는 대기 상태가 되기 때문이다.(이건 Kafka 동작의 근본과 관련된 것으로 동일한 파티션 내에서는 메시지의 순서를 보장해야 하기 때문이다.)

결과적으로 concurrency=2라고 해도 concurrency=1과 동일한 방식으로 메시지가 처리되었다.

3. 토픽 1개(파티션 3개, concurrency 6)

Received message in group 'test': Test Message 0 | Processed by: Thread-1
Received message in group 'test': Test Message 1 | Processed by: Thread-2
Received message in group 'test': Test Message 2 | Processed by: Thread-3

이번에는 파티션을 3개로 늘리고, concurrency 값을 6으로 설정해 보았다. Kafka는 파티션 단위로 컨슈머를 할당하기 때문에, 이 경우 서로 다른 파티션이기에 최대 3개의 스레드가 동시에 메시지를 소비할 수 있다. 실행 결과를 보면 3개의 스레드가 각각 다른 파티션에서 메시지를 병렬로 처리하고 있음을 확인할 수 있다. 하지만 concurrency 값을 6으로 설정했음에도 3개의 스레드만 사용되었는데, Kafka에서 활성화되는 컨슈머 스레드 개수는 할당된 파티션 개수를 초과할 수 없기 때문이다.

 

 

특정 토픽-파티션을 명시적으로 지정했을 때 vs 지정하지 않았을 때

여기서 조금 더 딥 다이브를 해보면 사실 정확히는 토픽-파티션을 명시적으로 지정했는지 여부에 따라 concurrency 스레드의 생성 방식은 달라진다.

1. 특정 토픽-파티션을 명시적으로 지정한 경우

특정 토픽-파티션을 명시적으로 지정한 경우

위를 보면 @KafkaListener에서 명시적으로 토픽-파티션 개수를 지정해 줬다. 이럴 경우 concurrency를 6으로 설정하더라도 Kafka는 애초에 지정된 파티션 개수까지만 생성한다. 따라서 실제로 실행되는 컨슈머 스레드는 3개이며 불필요한 추가 스레드는 생성되지 않는다.

2. 토픽-파티션을 명시적으로 지정하지 않았을 경우

토픽-파티션을 명시적으로 지정하지 않았을 경우

반면 특정 파티션을 명시적으로 지정하지 않은 경우 concurrency=6으로 설정하면 Spring Kafka는 6개의 컨슈머 스레드를 생성한다. 그리고 Kafka는 파티션 개수에 맞춰 컨슈머를 할당하기 때문에 실제로 메시지를 소비하는 스레드는 3개뿐이며, 나머지 3개는 유후상태로 남게 된다. 즉 6개의 스레드가 생성되지만 실제로 메시지를 소비하는 스레드는 3개뿐이라는 말이다.

실제 내부 구현 코드는 아래의 ConcurrentMessageListenerContainer.java#L243를 보면 된다.

https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java#L243

뭐 일반적으로 실제 현업에서는 일반적으로 파티션 개수를 명시적으로 지정하는 경우는 아마 잘 없을 것이다. 보통 토픽만 지정하고 특정 파티션까지는 지정하지 않으니까.. 그냥 알아두면 좋을 내용이다.

 

 

정리

이번 글에서는 Kafka의 @KafkaListener에서 concurrency 값을 활용한 병렬 처리 방식을 살펴보았다. concurrency는 Kafka에서 병렬 처리를 가능하게 하는 유용한 설정이지만, 파티션 개수를 초과하는 concurrency 값은 의미가 없으며, 설정 방식에 따라 생성되는 컨슈머 스레드 개수가 달라질 수 있음을 확인했다.

특히 특정 파티션을 명시적으로 지정한 경우에는 Kafka가 해당 파티션 개수까지만 컨슈머를 생성하지만, 그렇지 않으면 concurrency 값만큼 컨슈머 스레드가 생성되나 일부는 유휴 상태로 남았다.

결국 효율적인 병렬 처리를 위해서는 concurrency 값을 파티션 개수와 맞추는 것이 중요하다. 그리고 실무에 적용할 때는 고려해야 할 점이 있다. 바로 리밸런싱이 일어날 때인데 리밸런싱이 발생하면 파티션 재할당이 일어나고 이때 각 컨슈머가 항상 동일한 개수의 파티션을 할당받는다는 보장이 없으므로, 유휴 상태의 스레드가 발생할 가능성이 있다. 그래서 실무에 적용한다면 운영 환경과 시스템의 특성에 맞춰 유연하게 조정해야 함을 인지하자.