Kafka 메시지 처리 성능 높이기 (병렬 처리)

문제: Consumer가 메시지를 한 번에 하나씩만 처리하는 현상

API 요청을 3번 연속으로 보내면, sleep 3초를 걸어놨기 때문에 Consumer에 로그가 3초 간격으로 총 9초가 걸린다.

 

Consumer 역할을 하는 Spring Boot는 멀티 쓰레드 기반으로 여러 개의 요청을 병렬 처리할 수 있는 구조임에도,

왜 비효율적으로 요청을 하나씩 처리하고 있는 걸까?

이 문제의 원인은 Kafka의 중요한 개념인 Partition과 밀접한 관련이 있다.

---

파티션(Partition)

파티션(Partition)이란?

큐를 여러개로 늘려서 병렬 처리를 가능하게 하는 기본 단위

메시지를 순차 처리하는 것보다 병렬 처리하는 것이 훨씬 빠르므로, 파티션은 메시지 처리량에 큰 영향을 미치는 핵심 요인이다.

 

파티션의 특징

1. 각 토픽은 하나 이상의 파티션으로 구성할 수 있다.

토픽을 생성할 때 별도의 옵션을 주지 않으면 아래와 같이 파티션을 1개만 생성한다.

하지만 아래와 같이 토픽을 생성할 때 파티션을 여러개 만들 수도 있다.

2. Producer가 특정 토픽에 메시지를 넣으면, 여러 파티션에 메시지가 적절하게 분산된다.

파티션이 여러개인 토픽에 메시지를 넣으면, 여러 파티션에 메시지가 적절하게 분산된다.

3. 하나의 파티션은 하나의 컨슈머에게만 할당된다.

하나의 파티션은 동일한 Consumer Group 내에서 단 하나의 컨슈머에게만 할당된다.

 

[정상 구조]

실제로 두 대의 컨슈머 서버를 실행시키고 여러번 요청을 보내보면, 한 서버에만 메시지 처리 로그가 찍힘을 확인할 수 있다.

[잘못된 구조]

여러 컨슈머가 하나의 파티션의 메시지를 같이 처리할 수는 없다.

4. 하나의 컨슈머가 여러 파티션을 처리할 수 있다.

여러 컨슈머가 하나의 파티션의 메시지를 같이 처리할 수는 없지만, 하나의 컨슈머가 여러 파티션을 처리하는 건 가능하다.

5. 하나의 파티션에 할당된 하나의 컨슈머는 메시지를 순서대로 처리한다.

파티션 단위로 메시지의 처리 순서를 보장하기 위해, 오프셋이 0인 메시지와 1인 메시지를 병렬적으로 처리하지 않는다.

컨슈머 #0 은 오프셋이 0인 메시지를 다 처리하고 난 뒤에 오프셋이 1인 메시지를 처리한다

이 특징 때문에 Spring Boot가 멀티 쓰레드를 기반으로 여러 개의 요청을 처리할 수 있는 구조임에도 불구하고,
Consumer가 이메일 발송 작업을 한 번에 하나씩만 처리하고 있었던 것이다.

---

파티션 설정

특정 토픽의 파티션 수 조회하기

토픽의 세부 정보를 조회하면 나온다.

PartitionCount : 토픽이 가지고 있는 파티션의 총 개수 Partition : 파티션 번호 (0번 부터 시작)

 

토픽 생성할 때 파티션 수 설정하기

# 문법
$ bin/kafka-topics.sh --bootstrap-server <kafka 주소> \
	--create --topic <토픽명> --partitions <파티션 수>
	
# 예제
$ bin/kafka-topics.sh --bootstrap-server localhost:9092 \
	--create --topic test.topic --partitions 3

 

 

기존 토픽의 파티션 수 늘리기

# 문법
$ bin/kafka-topics.sh --bootstrap-server <kafka 주소> \
	--alter --topic <토픽명> --partitions <변경할 최종 파티션 수>
	
# 예제
$ bin/kafka-topics.sh --bootstrap-server localhost:9092 \
	--alter --topic test.topic --partitions 5

 

기존 토픽의 파티션 수 줄이기

Kafka에서는 파티션 수를 늘릴 수는 있지만 줄일 수는 없게 만들어놨다. 파티션을 줄이는 과정에서 내부적으로 문제(데이터 손실, 성능 저하 등)가 많이 발생하기 때문이다.

그래서 파티션을 줄이고 싶다면 새로운 토픽을 생성해서 파티션 수를 다시 설정하고, 기존 토픽을 새로운 토픽으로 마이그레이션 시켜야 한다. (이 과정은 꽤 귀찮기 때문에, 처음 파티션 수를 설정할 때 신중하게 설정하자)

---

실습

여러 개의 파티션에 메시지가 골고루 들어가는 지 확인해보기

특정 토픽에 메시지를 넣으면 여러 파티션에 메시지가 적절하게 분산된다고 했다.

이 때, 메시지의 형태에 따라 파티션에 분배되는 방식이 달라진다.

1. key가 포함되지 않은 메시지를 넣을 경우
   스티키 파티셔닝 방식으로 메시지를 분배(default)
   - 성능 상 배치 단위로 처리하기 위해 하나의 파티션에 메시지가 일정량이 채워져야만 그 다음 파티션에 메시지를 저장
   - 스티키 파티셔닝 방식이 대규모의 데이터를 처리할 때는 유리하지만, 작은 규모의 데이터를 처리할 때는 하나의 파티션에만 메시지가 몰리기 때문에 비효율적
   - Producer의 설정을 변경하면 라운드 로빈 방식 적용 가능(번갈아가며 차례대로 하나씩 배분)
2. key가 포함된 메시지를 넣을 경우
   key의 해시 값을 기반으로 파티션을 결정해서 메시지를 분배. (같은 key 값을 가진 메시지는 같은 파티션에 들어간다)

확인을 위해 프로듀서 서버의 applicaition.yml에 라운드 로빈 방식으로 설정 변경 후 확인해보자.

spring:
  kafka:
    bootstrap-servers: 52.79.47.232:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      # 라운드로빈 파티셔너 사용 (동일한 key를 가진 메시지들이 여러 파티션에 분산되어 저장되도록 함)
      properties:
        partitioner.class: org.apache.kafka.clients.producer.RoundRobinPartitioner

# email.send 토픽의 모든 메시지를 조회 (+ 파티션 정보도 같이 출력)
$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
	--topic email.send --from-beginning \
	--property print.partition=true

---

Spring Boot에서 여러 개의 컨슈머로 메시지 병렬적으로 처리하기

컨슈머 서버를 띄워보면 하나의 컨슈머 서버가 파티션 3개에 할당되어 있는 것을 확인할 수 있는데,

실제 요청을 보내보면 컨슈머 서버가 한 번에 하나의 작업만 처리하는 걸 확인할 수 있다.

여러 작업을 병렬적으로 처리하지 않으니 3개의 메시지를 처리하는 데 비교적 시간이 오래걸린다.

 

먼저 컨슈머 서버를 2대 띄워 아래 구조로 여러 컨슈머를 이용해 메시지를 병렬 처리해보자.

 

요청을 보내보면 2개의 컨슈머 서버가 토픽의 메시지를 병렬적으로 처리하는 걸 확인할 수 있다.

컨슈머 서버가 하나일 때보다 훨씬 효율적으로 처리한다.

 

하지만 여기서 아쉬운 점이 있다.
컨슈머 서버가 메시지를 처리할 때 사용하는 리소스(CPU, 메모리 등)가 부족한 상태라면 다른 컨슈머 서버를 늘리는 게 맞다.
하지만 이번 실습에서는 메일 작업 하나가 무거운 작업도 아닐 뿐더러, 하나의 컨슈머 서버의 리소스가 부족한 상태도 아니었다. 그래서 다음에는 컨슈머 서버를 늘리지 않고 여러 파티션의 메시지를 병렬적으로 처리하는 방법에 대해 알아보자.

 

Spring Boot 기반 하나의 컨슈머로 메시지 병렬적으로 처리하기

왜 하나의 컨슈머에 파티션을 3개 할당해도 병렬 처리하지 않고 순차 처리할까? 스프링 부트는 멀티스레드 기반인데 말이다.

Kafka “컨슈머 1개 인스턴스”는 기본적으로 “1개 스레드”로 poll → 처리 루프를 돌기 때문이다.

 

Spring Boot 앱 전체는 멀티스레드가 맞지만, 

@KafkaListener는 내부적으로 KafkaMessageListenerContainer라는 “리스너 컨테이너”에서 돌고, 이 컨테이너는 기본적으로 스레드 1개로 동작한다.

 

따라서 컨슈머 인스턴스를 3개 띄우던지, 리스너 컨테이너를 3개 만들어 병렬 소비하게 하면 된다.

이번에는 후자를 적용해보자.

@KafkaListener(topics = "email.send", groupId = "email-send-group", concurrency = "3")
public void consume(String message) { ... }

 

위와 같이 적용 후 API 요청을 3개 보내면 3개가 동시에 처리된다.

하나의 컨슈머 서버에서 하나의 쓰레드가 하나의 파티션에 있는 메시지를 처리하는 방식

다만 스레드가 3개이므로, 5개의 요청을 동시에 보내면 3개가 먼저 실행되고, 처리되는 대로 나머지가 실행된다.

해당 토픽의 파티션이 3개이므로 여러 쓰레드가 병렬적으로 처리할 수 있는 메시지의 개수가 최대 3개일 수 밖에 없기 때문이다.

그렇다고 파티션 수를 무작정 과도하게 늘리게 되면 오히려 반대로 성능에 비효율성을 가져온다.

그래서 적절한 파티션 수를 설정하는 게 중요하다.

---

적정 파티션 개수 계산하는 방법

적정 파티션 개수를 정할 때의 핵심은 ‘처리가 지연되는 메시지가 생기지 않는 선에서 파티션을 최소로 설정하는 것’이다.

공식으로 표현하자면 다음 공식을 만족하게 파티션 수를 결정하면 된다.

프로듀서가 보내는 메시지량 ≤ 하나의 쓰레드가 처리하는 메시지량 x 파티션 수

 

트래픽과 한 쓰레드 당 처리량(Throughput)은 부하테스트로 측정 가능하므로, 파티션 수를 계산할 수 있다.

트래픽이 100, 한 쓰레드 당 처리량(1초에 처리할 수 있는 양)이 0.3이라고 하면,
대략 트래픽을 널널하게 120으로 잡고 120 = 0.3 *  400 으로 파티션 수를 구할 수 있는 것이다.

부하 테스트로 측정하기
1. 몇 개의 쓰레드를 사용해야 처리량이 가장 높아지는 지 측정
Spring Boot 서버는 멀티 쓰레드 기반이기 때문에 동시에 여러 요청을 처리할 수 있다. 이 때, 몇 개의 쓰레드를 사용해야 요청을 가장 많이 처리할 수 있는 지 측정해야 한다.
(100개의 쓰레드를 활용하는게 가장 효율적으로 측정했다고 가정해보자)

2. 하나의 컨슈머 서버가 처리할 수 있는 최대 처리량(Throughput) 측정
컨슈머 서버가 적절한 쓰레드 개수를 기반으로 요청을 처리한다고 했을 때, 최대 처리량(Throughput)이 얼마나 되는 지 측정해야 한다.
(하나의 컨슈머 서버(100개의 쓰레드 활용)가 1초에 처리할 수 있는 처리량이 30이라고 가정 -> 쓰레드 1개가 초당 0.3개)

3. 프로듀서가 보내는 평균 메시지량 측정 or 예상
(사용자가 평균적으로 1초당 100개의 메일을 보낸다고 가정) 

---

컨슈머가 메시지를 지연없이 잘 처리하고 있는 지 확인하는 방법 (Consumer Lag)

Lag 이란?

평소에 컴퓨터가 느려지거나 버벅거리면 우리는 “렉 걸린다”라는 표현을 쓰는데, 이는 영어 단어인 Lag을 한글로 읽은 것이다.

Lag의 뜻은 ‘지연, delay’의 의미를 가진다. 

 

카프카에서 사용하는 랙(Lag)의 의미는, 지연된 메시지 수(컨슈머가 아직 처리하지 못한 메시지 수)이다. Consumer Lag이라고도 부른다.

 

컨슈머 랙(Consumer Lag)은 언제 발생할까?

프로듀서의 메시지 생산량보다 컨슈머의 메시지 처리량이 작을 때 컨슈머 랙(Consumer Lag)이 발생한다.

메시지가 1초에 3개씩 생기는데 1초에 메시지를 1개씩 밖에 처리를 못한다면, 1초 당 2개의 메시지가 계속해서 쌓일 것이다.

 

실무에서는 보통 갑작스럽게 요청이 증가할 때나 컨슈머에 장애가 생겼을 때 컨슈머 랙이 발생한다.

컨슈머 랙이 발생했다는 건 메시지 처리가 지연되고 있다는 뜻이다. 따라서 컨슈머 랙이 시간이 갈수록 계속해서 늘어나고 있다면 빠르게 조치를 취해야 한다. 빠르게 조치를 취하려면 컨슈머 랙을 지속적으로 모니터링 할 수 있어야 한다.

 

컨슈머 랙(Consumer Lag) 확인하는 방법

CLI로 컨슈머 랙 확인 방법

프로듀서 서버만 켜두고, 컨슈머 서버 종료하면 컨슈머에 장애가 걸린 상황을 만들 수 있다.

컨슈머 그룹의 세부 정보를 조회하면 컨슈머 랙에 대한 정보가 나온다.

 

하지만 매번 컴퓨터 앞에 앉아서 CLI로 컨슈머 랙이 생기는 지 안 생기는 지를 24시간 동안 체크할 수는 없다.

 

현업에서의 컨슈머 랙(Consumer Lag) 체크 방법

1. 외부 모니터링 툴 사용
   외부 모니터링 툴을 사용해 Conusmer Lag을 지속적으로 모니터링하면서, 특정 케이스에 대해 알림을 발송하게 만들어서 빠르게 대처할 수 있게 셋팅하는 편이다.
   Datadog, Burrow, Prometheus, Grafana 등을 주로 사용한다.
2. 매니지드 서비스(Managed Service)에서 제공하는 모니터링 기능 사용
   현업에서는 카프카를 직접 구축해서 사용하지 않고, 클라우드의 카프카 서비스를 사용하는 경우도 많다.
   대표적인 서비스로 AWS MSK와 Confluent Cloud가 있다. 이 서비스를 사용하면 자체적으로 Consumer Lag에 대한 모니터링 기능을 같이 제공하는 경우가 많다. 

---

 

참고 자료 & 이미지 출처
실전에서 바로 써먹는 kafka 입문