카프카 기본 개념과 구조
컨슈머의 기본 동작과 예제 맛보기
컨슈머는 카프카의 토픽에 저장되어 있는 메세지를 가져오는 역할을 담당한다
컨슈머의 기본 동작
- 프로듀서가 토픽으로 메세지를 전송하면 메세지들은 브로커들의 로컬 디스크에 저장된다
- 컨슈머를 이용해 토픽에 저장된 메세지를 가져올 수 있다
- 컨슈머 그룹은 하나 이상의 컨슈머들이 모여있는 그룹을 의미한다
- 컨슈머는 반드시 컨슈머 그룹에 속하게 된다
- 컨슈머 그룹은 각 파티션의 리더에게 카프카 토픽에 저장된 메세지를 가져오기 위한 요청을 한다
- 파티션 수와 컨슈머 수는 일대일로 매핑되는 것이 이상적이다
- 파티션 수보다 컨슈머 수가 많게 구현하는 것은 바람직한 구현이 아니다
- 컨슈머가 많다고 처리량이 높아지는 것이 아니라, 더 많은 수의 컨슈머들이 대기 상태로만 존재하기 때문
- 액티브/스탠바이 개념으로 추가 컨슈머가 있으면 좋을 것이라 생각할 수도 있지만, 컨슈머 그룹 내에서 리밸런싱 동작을 통해 장애가 발생한 컨슈머의 역할을 동일한 그룹에 있는 다른 컨슈머가 그 역할을 대신 수행하므로 굳이 장애 대비를 위한 추가 컨슈머 리소스를 할당하지 않아도 된다
컨슈머의 주요 옵션
| 컨슈머 옵션 | 설명 |
|---|---|
| bootstrap.servers | 프로듀서와 동일하게 브로커의 정보를 입력 |
| fetch.min.bytes | 한 번에 가져올 수 있는 최소 데이터 크기 지정된 크기보다 작은 경우, 요청에 응답하지 않고 데이터가 누적될 때까지 기다린다 |
| group.id | 컨슈머가 속한 컨슈머 그룹을 식별하는 식별자 |
| heartbeat.interval.ms | 하트비트가 있다는 것은 컨슈머의 상태가 active임을 의미한다. session.timeout.ms와 밀접한 관계가 있으며 session.timeout.ms 보다 낮은 값으로 설정해야 한다 일반적으로 session.timeout.ms 의 1/3으로 설정한다 |
| max.partition.fetch.bytes | 파티션당 가져올 수 있는 최대 크기를 의미한다 |
| enalbe.auto.commit | 백그라운드로 주기적으로 오프셋을 커밋한다 |
| auto.offset.reset | 카프카에서 초기 오프셋이 없거나 현재 오프셋이 더 이상 존재하지 않는 경우에 다음 옵션으로 reset 한다 *earliest: 가장 초기의 오프셋 값으로 설정한다 *latest: 가장 마지막의 오프셋 값으로 설정한다 *none: 이전 오프셋 값을 찾지 못하면 에러를 나타낸다 |
| fetch.max.bytes | 한 번의 가져오기 요청으로 가져올 수 있는 최대 크기 |
| group.instance.id | 컨슈머의 고유한 식별자 만약 설정한다면 static 멤버로 간주되어, 불필요한 리밸런싱을 하지 않는다 |
| isolation.level | 트랜잭션 컨슈머에서 사용되는 옵션으로, read_uncommitted는 기본적으로 모든 메세지를 읽고, read_committed는 트랜잭션이 완료된 메세지만 읽는다 |
| max.poll.records | 한 번의 poll() 요청으로 가져오는 최대 메세지 수 |
| partition.assignment.strategy | 파티션 할당 전략, 기본 값은 range |
| fetch.max.wait.ms | fetch.min.bytes 에 의해 설정된 데이터보다 적은 경우 요청에 대한 응답을 기다리는 최대 시간 |
컨슈머 예제
컨슈머 그룹의 이해
컨슈머 그룹이란?
컨슈머는 컨슈머 그룹 안에 속한 것이 일반적인 구조로, 하나의 컨슈머 그룹 안에 여러개의 컨슈머가 구성될 수 있다. 일반적으로 토픽의 파티션과 일대일로 매핑되어 메세지를 가져오게 된다
- 컨슈머들은 하나의 컨슈머 그룹 안에 속해 있다
- 그룹 내의 컨슈머 들은 서로의 정보를 공유한다
- 컨슈머01이 문제가 생겨 종료됐다면, 컨슈머02 또는 컨슈머03은 컨슈머01이 하던 일을 대신해 peter-01 토픽의 파티션0을 컨슘한다
우렁총각