[MSA] kafka - consumer

📌 개인적인 공간으로 공부를 기록하고 복습하기 위해 사용하는 블로그입니다.
정확하지 않은 정보가 있을 수 있으니 참고바랍니다 :😸
[틀린 내용은 댓글로 남겨주시면 복받으실거에요]

인프런 Dowon Lee님의 Spring Cloud로 개발하는 마이크로서비스 애플리케이션(MSA) 강의를 듣고 정리한 내용입니다.😊
Spring Cloud로 개발하는 마이크로서비스 애플리케이션(MSA) 강의 들으러 가기👩‍🏫
✅ 추가로 kafka 에 대해 잘 정리된 블로그가 있어 참고 하였습니다. » 언제나김김

카프카 컨슈머(Kafka Consumer)

• 카프카 메시지 시스템에서 데이터를 소비하는 역할을 담당
• 프로듀서가 브로커에 데이터를 전송하면, 컨슈머는 해당 데이터를 읽어와 애플리케이션에서 처리한다.

1️⃣ 컨슈머 구조

컨슈머 그룹(Consumer Group)

• 컨슈머 그룹은 하나 이상의 컨슈머로 구성되며, 동일한 그룹에 속한 컨슈머들은 토픽의 파티션들을 분할하여 처리한다.
• 특징
  • 격리성: 서로 다른 컨슈머 그룹은 독립적으로 동작하며, 동일한 토픽을 구독하더라도 서로 영향을 주지 않는다.
  • 파티션 할당: 하나의 파티션은 동일한 컨슈머 그룹 내에서 단 하나의 컨슈머에게만 할당된다.
  • 컨슈머 수와 파티션 수: 컨슈머 그룹의 컨슈머 수는 토픽의 파티션 수보다 많을 필요가 없다. 컨슈머 수가 파티션 수보다 많으면 남는 컨슈머는 데이터를 처리하지 않는다.

컨슈머와 파티션의 매핑

• N:1 관계: 하나의 컨슈머는 여러 개의 파티션을 처리할 수 있지만, 하나의 파티션은 한 컨슈머에게만 할당된다.
• 확장성: 컨슈머 수를 늘려 병렬 처리 성능을 향상시킬 수 있지만, 파티션 수 이상의 효과는 없다.

2️⃣ 오프셋(Offset)과 커밋(Commit)

오프셋(Offset)

• 컨슈머가 파티션에서 읽은 메시지의 위치를 나타내는 고유한 번호
• 컨슈머는 오프셋을 기반으로 어디까지 데이터를 읽었는지 추적하고, 재시작 시 이어서 처리할 수 있다.

오프셋 커밋(Commit)

• 컨슈머가 처리한 마지막 오프셋을 브로커에 저장하여, 재시작 시 해당 위치부터 데이터를 읽을 수 있도록 한다.
• 저장 위치: 카프카 내부 토픽인 __consumer_offsets에 저장된다.
• 커밋 방식
  • 자동 커밋(Auto Commit):
    • 설정: enable.auto.commit=true
    • 컨슈머가 일정 주기(auto.commit.interval.ms)마다 자동으로 오프셋을 커밋한다.
    • 장점: 구현이 간단하며, 별도의 코드 작성이 필요 없다.
    • 단점: 메시지 처리 중단 시 데이터 중복 또는 유실 가능성이 있다.
  • 수동 커밋(Manual Commit):
    • 설정: enable.auto.commit=false
    • 방법
      • 동기 커밋(commitSync): 커밋 요청을 보내고 응답을 받을 때까지 기다림
      • 비동기 커밋(commitAsync): 커밋 요청을 보내고 바로 다음 작업을 수행한다.
    • 장점: 메시지 처리 완료 후 정확한 시점에 커밋할 수 있어 데이터 중복이나 유실을 최소화한다.
    • 단점: 구현이 복잡하며, 성능에 영향을 줄 수 있다.

데이터 중복과 유실

• 중복 발생 가능성: 컨슈머 장애나 리밸런싱 시 마지막으로 커밋된 오프셋 이후의 데이터를 다시 처리하게 되어 중복이 발생할 수 있다.
• 유실 발생 가능성: 자동 커밋 사용 시 메시지 처리 전에 오프셋이 커밋되면 처리되지 않은 데이터가 유실될 수 있다.
• 해결 방안:
  • 애플리케이션 레벨에서 중복 처리 로직을 구현
  • 멱등성을 보장하는 처리를 고려

3️⃣ 폴링(Polling) 메커니즘

Poll 메서드

• 동작 방식: 컨슈머는 poll() 메서드를 주기적으로 호출하여 브로커로부터 데이터를 가져옵니다.
• 내부 동작
  • 컨슈머는 백그라운드에서 Fetcher 스레드를 통해 데이터를 미리 가져와 내부 큐에 저장합니다.
  • poll() 호출 시 내부 큐에서 데이터를 반환한다.

하트비트(Heartbeat)

• 역할: 컨슈머는 주기적으로 브로커(코디네이터)에게 하트비트를 전송하여 자신의 상태를 알린다.
• 설정 옵션
  • heartbeat.interval.ms: 하트비트 전송 주기(기본값: 3초)
  • session.timeout.ms: 컨슈머 세션 타임아웃 시간(기본값: 10초)
    • 이 시간 내에 하트비트를 받지 못하면 브로커는 컨슈머가 비정상이라고 판단하고 리밸런싱을 수행한다.

폴링 주기와 리밸런싱

• 폴링 지연 시 문제점
  • max.poll.interval.ms 설정 시간(기본값: 5분) 내에 poll()을 호출하지 않으면 컨슈머는 비정상으로 판단된다.
  • 이 경우 리밸런싱이 발생하여 컨슈머 그룹 전체의 메시지 처리가 일시 중단될 수 있다.
• 해결 방안
  • max.poll.records를 조정하여 한 번에 가져오는 레코드 수를 줄인다.
  • 레코드 처리 시간을 최적화하여 poll() 호출 간격을 단축한다.
  • 스레드 분리 등을 통해 메시지 처리와 폴링을 병렬화한다.

4️⃣ 컨슈머 주요 옵션

필수 옵션

1. bootstrap.servers
   • 카프카 클러스터의 브로커 정보를 지정
2. key.deserializer
   • 메시지 키를 역직렬화하는 클래스
3. value.deserializer
   • 메시지 값을 역직렬화하는 클래스
4. group.id
   • 컨슈머 그룹의 아이디를 지정

선택 옵션

1. auto.offset.reset
   • 컨슈머 그룹의 오프셋이 없을 때 어디서부터 읽을지 지정한다.
   • 옵션 값:
     • latest(기본값): 가장 마지막 오프셋부터 읽는다.
     • earliest: 가장 처음 오프셋부터 읽는다.
     • none: 오프셋이 없으면 에러를 발생시킨다
2. enable.auto.commit
   • 자동 커밋 사용 여부를 결정
   • 기본값은 true
3. auto.commit.interval.ms
   • 자동 커밋일 경우 오프셋 커밋 주기를 지정
   • 기본값은 5,000ms(5초)
4. fetch.min.bytes
   • 브로커로부터 한 번에 가져올 최소 데이터 크기
   • 기본값은 1바이트
5. fetch.max.bytes
   • 브로커로부터 한 번에 가져올 수 있는 최대 데이터 크기
   • 기본값은 50MB
6. max.poll.records
   • 한 번의 poll() 호출로 가져올 수 있는 최대 레코드 수
   • 기본값은 500입
7. max.poll.interval.ms
   • poll() 호출 간 최대 지연 시간
   • 기본값은 300,000ms(5분)
8. session.timeout.ms
   • 컨슈머 세션 타임아웃 시간
   • 기본값은 10,000ms(10초)
9. heartbeat.interval.ms
   • 하트비트 전송 주기
   • 기본값은 3,000ms(3초)
10. isolation.level
    • 트랜잭션 처리를 위한 옵션
    • read_uncommitted(기본값), read_committed 중 선택

5️⃣파티션 할당 전략과 리밸런싱

파티션 할당 전략(Partition Assignment Strategy)

• 컨슈머 그룹 내에서 파티션을 컨슈머에게 어떻게 할당할지 결정
• 옵션:
  • RangeAssignor(기본값)
    • 토픽별로 파티션을 범위 단위로 나누어 컨슈머에게 할당
  • RoundRobinAssignor
    • 모든 파티션을 컨슈머들에게 순차적으로 할당
  • StickyAssignor:
    • 이전 할당 상태를 최대한 유지하며 파티션을 할당
  • CooperativeStickyAssignor
    • 협력적인 리밸런싱을 통해 최소한의 이동으로 파티션을 재분배

리밸런싱(Rebalancing)

• 발생 조건:
  • 컨슈머 그룹 내 컨슈머의 추가 또는 제거
  • 토픽의 파티션 수 변경
• 동작 과정:
  1. 컨슈머 그룹 내 컨슈머들은 poll() 호출 시 리밸런싱 필요 여부를 확인
  2. 리밸런싱이 필요하면 컨슈머들은 코디네이터에게 조인 요청을 보낸다
  3. 코디네이터는 컨슈머들의 조인 요청을 받은 후 그룹 리더를 선출
  4. 그룹 리더는 파티션 할당 전략에 따라 파티션을 컨슈머들에게 분배하고 코디네이터에게 전달한다
  5. 코디네이터는 각 컨슈머에게 새로운 파티션 할당 정보를 전송
• 주의사항:
  • 리밸런싱 동안 컨슈머들은 메시지를 처리하지 못하고 대기하게 된다
  • 리밸런싱 시간이 길어지면 메시지 처리 지연(Lag)이 발생할 수 있다.

리밸런싱 최적화 방안

1. 폴링 주기 단축
   • max.poll.records를 줄여 poll() 호출 빈도를 높인다.
2. 스레드 분리
   • 메시지 처리 로직을 별도의 스레드로 분리하여 poll() 호출 지연을 방지한다
   • 단, 파티션별 메시지 처리 순서를 보장하기 어렵다
3. Cooperative Rebalancing 사용
   • partition.assignment.strategy를 CooperativeStickyAssignor로 설정하여 리밸런싱의 영향을 최소화한다

컨슈머 애플리케이션의 일반적인 구조

• 단일 스레드 구조
  • poll() 메서드로 레코드를 가져오고, 같은 스레드에서 처리한다.
  • 구현이 간단하지만, 메시지 처리 시간이 길어지면 폴링 주기가 길어져 리밸런싱에 영향이 있다.
• 멀티 스레드 구조
  • 메시지 처리를 별도의 스레드 풀에서 수행
  • 폴링 주기를 짧게 유지할 수 있지만, 파티션별 메시지 순서가 보장되지 않을 수 있다.
  • 스레드 풀의 크기와 백프레셔(Backpressure) 관리가 필요하다.

데이터 처리 시 고려사항

• 메시지 순서 보장
  • 파티션 내 메시지 순서를 보장하려면 단일 스레드로 처리하거나, 순서 보장 로직을 구현해야 한다.
• 중복 처리 방지
  • 컨슈머 장애 시 중복 처리가 발생할 수 있으므로, 멱등성을 고려한 로직이 필요하다.
• 백프레셔 관리

  • 메시지 처리 속도와 수신 속도의 균형을 맞춰 메모리 이슈를 방지한다.

MSA 카테고리 내 다른 글 보러가기