카프카 클러스터에 명령을 내리거나 데이터를 송수신하기 위해 카프카 클라이언트 라이브러리는 카프카 프로듀서, 컨슈머, 어드민 클라이언트를 제공하는 카프카 클라이언트를 사용해서 애플리케이션을 개발한다.
프로듀서 API: 전송 시 리더 파티션을 가지고 있는 카프카 브로커와 통신, 데이터를 직렬화 하여 전송, 자바 기본형 데이터 뿐만 아니라 동영상, 이미지 같은 바이너리 데이터도 프로듀서를 통해 전송할 수 있다.
public class SimpleProducer {
private final static Logger logger = LoggerFactory.getLogger(SimpleProducer.class);
private final static String TOPIC_NAME = "test";
private final static String BOOTSTRAP_SERVERS = "my-kafka:9092";
public static void main(String[] args) {
Properties configs = new Properties();
configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
KafkaProducer<String, String> producer = new KafkaProducer<>(configs);
String messageValue = "testMessage";
ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC_NAME, messageValue);
producer.send(record);
producer.flush();
producer.close();
}
}
실제 운영 환경에서는 데이터가 정상적으로 전송되었는지 확인하는 로직, 환경에 따른 추가적인 프로듀서 선택 옵션들의 설정 등과 같은 로직들이 포함된다.
애플리케이션 단에서 토픽에 레코드를 전송하기 전에 카프카 커맨드를 이용해서 미리 토픽을 만들어준다.
프로듀서 애플리케이션은 카프카 클러스터로 레코드를 전송한다.
이 떄 내부적으로 파티셔너, 어큐물레이터, 센더를 거치게 된다.
레코드 생성 시 파티션을 직접 지정하거나 타임스탬프 지정, 메세지 키 지정 등이 가능하다.
프로듀서 인스턴스의 send()를 호출하게 되면 먼저 파티셔너에서 토픽의 어느 파티션으로 전송할 것인지 결정된다.
파티션이 구분된 데이터는 어큐물레이터에 토픽별로 버퍼에 쌓인 다음 배치 단위로 전송된다.
프로듀서 필수 옵션
bootstrap.servers: 프카프카 클러스터에 속한 ㅍ브로커의 호스트 이름:포트를 1개 이상 작성
key.serializer: 레코드의 메시지 키를 직렬화 하는 클래스를 지정
value.serializer: 레코드의 메시지 값을 직렬화하는 클래스 지정
프로듀서 선택 옵션
acks: 프로듀서가 전송한 데이터가 브로커들에 정상적으로 저장되었는지 전송 성공 여부를 확인하는데 사용
0, 1, -1 중 택 1, 1은 리더 파티션에 데이터가 저장되면 전송 성공, 0은 프로듀서가 전송한 즉시 브로커에 저장 여부와 상관없이 성공, -1은 리더 파티션과 팔로워 파티션에 데이터가 저장되면 성공하는 것으로 판단
buffer.memory: 브로커로 전송할 데이터를 배치로 모으기 위해 설정할 버퍼 메모리 양
retries: 브로커로부터 에러를 받고 난 뒤 재전송 시도 횟수 설정
batch.size: 배치로 전송할 레코드 최대 용량 지정, 기본값은 16384
linger.ms: 배치를 전송하기 전까지 기다리는 최소 시간, 기본값은 0
partitioner.class: 레코드를 파티션에 전송할 때 적용하는 파티션 클래스
enable.idempotence: 멱등성 프로듀서로 동작할 지 여부를 설정
transactional.id: 프로듀서가 레코드를 전송할 때 레코드를 트랜잭션 단위로 묶을지 여부를 설정
메시지 키를 가진 데이터 전송 프로듀서
// key 설정 레코드
ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC_NAME, "Pangyo", "23");
// partition 지정 레코드
ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC_NAME, 0, "Pangyo", "23");
커스텀 파티셔너
Partitioner 인터페이스를 직접 구현해서 특정 키가 특정 파티션에 배정되도록 적용할 수 있다.
브로커 정상 전송 여부 확인 프로듀서
ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC_NAME, messageValue);
RecordMetadata metadata = producer.send(record).get();
logger.info(metadata.toString());
producer.send(record).get() 함수를 호출하면 동기적으로 프로듀서가 보낸 데이터의 결과를 확인할 수 있다.
레코드가 정상적으로 브로커에 적재되면 토픽 이름-파티션번호-오프셋번호 가 리턴된다. ex) test-1@3
콜백 함수를 만든 뒤, send 함수에 콜백 함수를 추가해서 전송하면 비동기적으로 데이터 전송 결과를 확인할 수 있다.
카프카 클라이언트
프로듀서 필수 옵션
프로듀서 선택 옵션
메시지 키를 가진 데이터 전송 프로듀서
커스텀 파티셔너
브로커 정상 전송 여부 확인 프로듀서