[Kafka] 카프카 핵심 가이드 - 3 (카프카 프로듀서)

- 데이터를 쓸 때 사용하는 프로듀서의 디자인과 주요 요소의 전체적인 모습을 살펴본다.

- KafaProducer와 ProducerRecord 객체를 어떻게 생성하고 전송하는지, 에러 처리와 설정 옵션을 살펴본다.

- 파티션의 할당 방식을 정의하는 파티셔너와 객체의 직렬화 방식을 정의하는 것은 어떤 것이 있는 지 살펴본다.

 

Producer 개요

 

애플리케이션이 카프카에 메시지를 써야하는 상황들의 예시:

- 감사 혹은 분석을 목적으로 한 사용자 행동 기록

- 성능 메트릭 기록

- 로그 메시지 저장

- 스마트 가전에서의 정보 수집

- 다른 애플리케이션과의 비동기적 통신 수행

- 임의의 정보를 데이터베이스에 저장하기 전 버퍼링

 

요구조건의 예시:

- 모든 메시지가 중요해서 유실이 허용되면 안되는지? 혹은 유실 허용이 가능한지?

- 중복이 허용되도 상관없는지?

- 반드시 지켜야할 지연(latency)이나 처리율(throughtput)이 있는지?

 

 

 

1. ProducerRecord 객체를 생성. 레코드가 저장될 topic과 value 지정은 필수, key와 partition은 선택. 이 객체를 네트워크 상에서 전송될 수 있도록 직렬화해서 바이트 배열로 변환

2. 파티션을 명시적으로 지정하지 않은 경우, 해당 데이터를 파티셔너에게로 보냄. 파티셔너는 파티션을 결정하고 메시지가 전송될 토픽과 파티션이 확정되면 레코드를 레코드 배치(record batch)에 추가함

3. 별도의 스레드가 이 레코드 배치를 적절한 카프카 브로커에게 전송

4. 브로커가 메시지를 받으면 응답을 돌려주기 위해 레코드의 오프셋을 담은 RecoredMetadata 객체를 리턴함

 

 

카프카 프로듀서 생성

3개의 필수 속성값을 설정!

- bootstrap.servers: 브로커의 host:port 목록

- key.serializer: 레코드의 키의 값을 직렬화하기 위해 사용하는 시리얼라이저 클래스의 이름

- value.serializer: 레코드의 value 값을 직렬화하기 위해 사용하는 시리얼라이저 클래스 이름. 

 

*단, 최신 버전은 관련 클래스에 사전 정의된 정적 상수를 사용하는 방식

프로듀서: org.apache.kafka.clients.producer.ProducerConfig

컨슈머: org.apache.kafka.clients.consumer.ConsumerConfig

AdminClient: org.apache.kafka.clients.admin.AdminClientConfig

 

메시지 전송 방법

- Fire and Forget: 메시지를 서버에 전송만 하고 성공실패 여부는 신경쓰지 않음

- Synchronous send: 기술적으로 카프카 프로듀서는 언제나 비동기적이지만, 메시지 보내면 send() 메서드는 Future 객체를 리턴한다. 이때 다음 메시지를 저송하기 전 get() 메서드를 호출해서 실제 성공여부를 확인하는 작업이 있음

- Asynchronous send: 콜백 함수와 함께 send() 메서드를 호출하면 카프카 브로커로부터 응답을 받는 시점에서 자동으로 콜백함수가 호출됨

 

 

 

 

 

 

 

참고: kafka 조금 아는 척하기 2 (개발자용) - 프로듀서 https://www.youtube.com/watch?v=geMtm17ofPY