카프카 컨슈머 설정별 성능 측정해보기

Kafka Consumer 설정별 성능 측정 실험

1. 실험 개요

1.1 실험 목적

파티션 수를 늘리지 않고, Consumer 설정만으로 처리량을 얼마나 끌어올릴 수 있는지 측정

1.2 실험 전제 조건

• 파티션 수: 1개 고정 (파티션당 Consumer 1개, 1:1 매칭)
• 메시지 수: 각 패턴당 10,000개
• 처리 시간: 메시지당 10ms 고정 (Thread.sleep(10))
• 브로커/네트워크: 충분한 리소스 보장 (로컬 환경)
• 테스트 방식: 미리 10,000개 메시지를 토픽에 적재한 후, Consumer 시작하여 처리 속도 측정

1.3 테스트 환경

• Spring Boot: 3.x
• Spring Kafka: 3.2.x
• Kafka: 로컬 브로커
• Java: 17
• 실행 방식: 로컬에서 Kafka와 Spring Boot 애플리케이션 동시 실행

---

2. 성능에 영향을 주는 요소

2.1 고려한 설정

1. max.poll.records: 한 번의 poll()로 가져오는 메시지 개수
2. fetch.min.bytes / fetch.max.wait.ms: 네트워크 페치 관련 설정
3. 배치 리스너 vs 단건 리스너: 메시지 처리 단위
4. 워커 풀을 이용한 병렬 처리: Consumer 스레드 외부에서 처리

2.2 테스트에서 제외한 설정

• fetch.min.bytes, fetch.max.wait.ms:
  • 이미 메시지가 토픽에 쌓여있는 상황이므로 네트워크 대기 시간이 없음
  • 로컬 환경에서는 영향이 미미할 것으로 판단

---

3. 테스트 패턴 설계

총 5가지 패턴으로 실험을 구성했습니다.

패턴	리스너 타입	max.poll.records	처리 방식	예상 결과
패턴1	단건	10	동기 처리	가장 느림
패턴2	단건	500	동기 처리	느림
패턴3	배치	500	순차 처리	느림
패턴4	배치	500	병렬 처리 (워커풀 20)	빠름
패턴5	단건	500	비동기 처리 (@Async)	가장 빠름

---

4.1 패턴1: 단건 리스너 (max.poll.records=10)

특징: 가장 기본적인 Consumer 설정

@KafkaListener(
    groupId = "pattern1-group",
    topicPartitions = @TopicPartition(topic = "pattern1.events", partitions = {"0"}),
    properties = {"max.poll.records=10"}
)
public void listen(String message, Acknowledgment ack) {
    processMessage(message);  // Thread.sleep(10ms)
    ack.acknowledge();
}

설정 포인트:

• max.poll.records=10: 한 번에 10개만 가져옴
• 단건씩 동기 처리

---

4.3 패턴2: 단건 리스너 (max.poll.records=500)

특징: poll 개수만 늘린 패턴

@KafkaListener(
    groupId = "pattern2-group",
    topicPartitions = @TopicPartition(topic = "pattern2.events", partitions = {"0"}),
    properties = {"max.poll.records=500"}
)
public void listen(String message, Acknowledgment ack) {
    processMessage(message);  // Thread.sleep(10ms)
    ack.acknowledge();
}

설정 포인트:

• max.poll.records=500: 한 번에 500개 가져옴
• 여전히 단건씩 동기 처리 → 역시 처리 자체는 느림 로컬에서 해서 그런가 10개나 500개나 차이가 없었다..

---

4.4 패턴3: 배치 순차 처리 (max.poll.records=500)

특징: 배치로 받아서 순차 처리

@KafkaListener(
    groupId = "pattern3-group",
    containerFactory = "batchListenerContainerFactory",
    topicPartitions = @TopicPartition(topic = "pattern3.events", partitions = {"0"})
)
public void listen(List<String> messages, Acknowledgment ack) {
    for (String message : messages) {
        processMessage(message);  // Thread.sleep(10ms)
    }
    ack.acknowledge();
}

설정 포인트:

• 배치 리스너 사용
• 배치 내에서 순차 처리 → 병렬 처리 안 함
• 기본적 셋잉으로는 한 건이라도 실패하면 전체 배치 재처리하는 문제가 있음.

---

4.5 패턴4: 배치 병렬 처리 (워커 풀)

특징: 배치로 받아서 워커 풀에서 병렬 처리

private final ExecutorService executorService = Executors.newFixedThreadPool(20);

@KafkaListener(
    groupId = "pattern4-group",
    containerFactory = "batchListenerContainerFactory",
    topicPartitions = @TopicPartition(topic = "pattern4.events", partitions = {"0"})
)
public void listen(List<String> messages, Acknowledgment ack) {
    List<CompletableFuture<Void>> futures = messages.stream()
        .map(msg -> CompletableFuture.runAsync(() -> processMessage(msg), executorService))
        .collect(Collectors.toList());

    CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
    ack.acknowledge();
}

설정 포인트:

• 배치 리스너 + 워커 풀(20개 스레드)
• CompletableFuture를 이용한 병렬 처리
• 모든 메시지 처리 완료 후 ACK

---

4.6 패턴5: 단건 비동기 처리 (@Async)

특징: 단건으로 받아서 즉시 비동기 처리

@KafkaListener(
    groupId = "pattern5-group",
    topicPartitions = @TopicPartition(topic = "pattern5.events", partitions = {"0"}),
    properties = {"max.poll.records=500"}
)
public void listen(String message, Acknowledgment ack) {
    longTaskService.processPattern5(message)
        .thenCompose(result -> {
            ack.acknowledge();
            return CompletableFuture.completedFuture(result);
        });
}

@Async("kafkaAsyncExecutor")
public CompletableFuture<String> processPattern5(String message) {
    Thread.sleep(10);  // 10ms
    return CompletableFuture.completedFuture(message);
}

비동기 ThreadPool 설정:

@Configuration
public class AsyncConfig {
    @Bean
    public ThreadPoolTaskExecutor kafkaAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(20);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("kafka-async-");
        executor.initialize();
        return executor;
    }
}

설정 포인트:

• 단건 리스너 + @Async 처리
• Consumer 스레드는 즉시 다음 poll() 수행
• ThreadPool(core:10, max:50)에서 비동기 처리
• 거부 정책: Queue가 가득 차면 호출자 스레드에서 실행(CallerRunsPolicy)으로 설정

---

5. 실험 결과

5.1 정상 처리 (실패율 0%)

패턴	처리 시간	처리량 (msg/s)	순위
패턴5: 단건 비동기	5초	2,000.00	🥇 1위
패턴4: 배치 병렬	8초	1,250.00	🥈 2위
패턴3: 배치 순차	118초	84.75	🥉 3위
패턴2: 단건(poll=500)	117초	85.47	4위
패턴1: 단건(poll=10)	122초	81.97	5위

5.2 상세 통계

【패턴5: 단건비동기(poll=500)】
  수신: 10,000
  성공: 10,000 (100.0%)
  실패: 0 (0.0%)
  진행중: 0
  진행률: 100.0%
  처리시간: 5초
  처리량: 2000.00 msg/s
  완료여부: ✅ 완료

【패턴4: 배치병렬(poll=500)】
  수신: 10,000
  성공: 10,000 (100.0%)
  실패: 0 (0.0%)
  진행중: 0
  진행률: 100.0%
  처리시간: 8초
  처리량: 1250.00 msg/s
  완료여부: ✅ 완료

【패턴3: 배치순차(poll=500)】
  수신: 10,000
  성공: 10,000 (100.0%)
  실패: 0 (0.0%)
  진행중: 0
  진행률: 100.0%
  처리시간: 118초
  처리량: 84.75 msg/s
  완료여부: ✅ 완료

【패턴2: 단건(poll=500)】
  수신: 10,000
  성공: 10,000 (100.0%)
  실패: 0 (0.0%)
  진행중: 0
  진행률: 100.0%
  처리시간: 117초
  처리량: 85.47 msg/s
  완료여부: ✅ 완료

【패턴1: 단건(poll=10)】
  수신: 10,000
  성공: 10,000 (100.0%)
  실패: 0 (0.0%)
  진행중: 0
  진행률: 100.0%
  처리시간: 122초
  처리량: 81.97 msg/s
  완료여부: ✅ 완료

---

6. 결과 분석

6.1 패턴5가 가장 빠른 이유

단건 비동기 (2,000 msg/s) > 배치 병렬 (1,250 msg/s)

1. poll 오버헤드 최소화:
   • Consumer 스레드가 메시지를 받자마자 즉시 비동기 처리 후 다음 poll() 수행
   • max.poll.records=500이므로 한 번에 500개씩 가져와서 빠르게 비동기로 처리
   • 그리고 그냥 비동기로 넣으면서 바로 처리 했다고 판단하게끔 코드를 작성해서.. 제일 빠른걸로 보여지는데 바로 처리 안했다고 해도 가장 빠를것으로 예상됩니다,,
2. ThreadPool 효율성:
   • @Async ThreadPool (core:10, max:20)이 배치 워커풀(20)보다 유연하게 확장
   • Queue(100)를 통해 버퍼링 가능
3. ACK 처리 방식:
   • 배치 병렬: 전체 배치 완료 대기 → ACK
   • 단건 비동기: 개별 완료 즉시 ACK → poll() 빠르게 재개

6.2 배치 병렬이 2위인 이유

배치 병렬 (1,250 msg/s) > 배치 순차 (84.75 msg/s)

1. 병렬 처리의 효과:
   • 워커 풀(20개)로 동시 처리
   • 500개를 20개 스레드로 나눠서 처리 → 약 25개씩 처리
2. 배치 처리의 장점:
   • 한 번에 많은 메시지를 가져와서 처리
   • ACK 횟수 감소

6.3 패턴1~3이 느린 이유

단건 동기 처리의 한계

1. 패턴1, 2, 3 모두 동기 처리:
   • Consumer 스레드가 메시지 처리 완료까지 블로킹
   • 10ms × 10,000개 = 100초 (이론적 최소 시간)
   • 실제로는 120초 정도 소요 (오버헤드 포함)
2. max.poll.records의 영향이 미미:
   • 패턴1(poll=10): 122초
   • 패턴2(poll=500): 117초
   • 단 5초 차이 → 동기 처리에서는 poll 개수가 큰 영향 없음 로컬에서 5초니까 실제 운영환경에선 좀 더 차이가 있을 순 있을것 같음.
3. 배치 순차 처리도 동일:
   • 패턴3: 118초
   • 배치로 받아도 순차 처리하면 단건 동기와 동일

---

7. 실패 처리에 대한 고려사항

7.1 배치 리스너의 단점

배치 리스너는 하나라도 실패하면 전체 배치 재처리되는 특성이 있습니다. -> 물론 변경할 수 있긴한데 이ㅔㄱ 기본값입니다.

// 패턴3, 4 공통
for (String message : messages) {
    processMessage(message);  // 하나라도 예외 발생 시
}
// 전체 배치 재처리

실패율 10% 테스트 시 예상:

• 패턴3, 4: 배치 내 하나 실패 → 전체 재처리 → 처리 시간 증가
• 패턴5: 개별 실패 → 해당 메시지만 재처리 → 처리 시간 영향 적음

7.2 권장 사항

• 정상 처리가 대부분인 경우: 패턴5 (단건 비동기) 또는 패턴4 (배치 병렬)
• 실패율이 높은 경우: 패턴5 (단건 비동기) 권장
• 순서 보장이 필요한 경우: 패턴1~3 (동기 처리)

---

8. 결론

8.1 핵심 발견

1. max.poll.records는 동기 처리에서 거의 영향 없음
   • 패턴1(poll=10): 122초
   • 패턴2(poll=500): 117초
   • 단 5초 차이 (4% 개선)
2. 병렬 처리가 핵심
   • 동기 처리: ~85 msg/s
   • 배치 병렬: 1,250 msg/s (약 15배 향상)
   • 단건 비동기: 2,000 msg/s (약 24배 향상)
3. 파티션을 늘리지 않고도 24배 성능 향상 가능
   • 기본 설정 대비 2,000% 개선
   • 단순히 비동기 처리만으로도 큰 효과

8.2 최종 권장 패턴

상황별 최적 패턴:

상황	권장 패턴	이유
높은 처리량 필요	패턴5 (단건 비동기)	가장 빠름 (2,000 msg/s)
실패율이 높은 경우	패턴5 (단건 비동기)	개별 재처리로 영향 최소화
순서 보장 필요	패턴1~3 (동기 처리)	순차 처리 보장
배치 작업 필요	패턴4 (배치 병렬)	배치 단위 처리 + 높은 성능

8.3 실무 적용 시 고려사항

1. ThreadPool 크기 조정:
   • 패턴5: @Async ThreadPool (core:10, max:20, queue:100)
   • 패턴4: ExecutorService (fixed:20)
   • 메시지 처리 시간과 양에 따라 조정 필요
2. 메모리 관리:
   • max.poll.records=500 × 여러 Consumer = 메모리 사용량 증가
   • 적절한 값 설정 필요
3. 모니터링:
   • Consumer Lag 모니터링
   • ThreadPool 사용률 모니터링
   • 처리 실패율 모니터링

---

결론: 파티션 수를 늘리지 않고도, 비동기 처리 방식을 적용하면 24배의 성능 향상을 얻을 수 있었습니다. 

---

10. 참고 자료

• Spring Kafka Documentation
• Apache Kafka Consumer Configuration
• Spring @Async Documentation