카프카 프로듀서 기본 개요

Kafka Producer

Serializer: 메시지를 byte 배열로 변환

Partitioner: 어느 토픽의 파티션으로 보낼지 결정

버퍼&배치: 메시지 모음

sender: 배치를 전송

Producer 가 메시지를 보내게 되면 Serializer 를 통해 메시지를 byte 배열로 변환하고 Partitioner 를 통해 어느 토픽의 파티션으로 보낼지 결정한다. 결정된 메시지는 버퍼의 배치로 들어가게 되고 sender를 통해 kafka broker 에게 메시지를 전달하게 된다. Send와 Sender 는 각각 별도의 쓰레드로 동작하므로 Sender 가 메시지를 보내는 동안 배치로 메시지를 모으게 되고 또 배치가 찼는지, 배치의 갯수와 상관없이 차례대로 브로커에게 메시지를 전송하게 된다.

전송 결과

전송 결과 확인하지 않고 보내기


producer.send(new ProducerRecord<>("topic","value"));

전송 실패를 알 수 없다.

실패에 대한 별도 처리가 필요 없는 메시지 전송에 사용한다.

전송 결과 확인: Future


package java.util.concurrent;

public interface Future<V> {
  boolean cancel(boolean mayInterruptIfRunning);
  boolean isCancelled();
  boolean isDone();
  V get() throws InterruptedException, ExecutionException;
  V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException; }

사용 예시


Future<RecordMetadata> f = producer.send(new ProducerRecord<>("topic","value"));
try {
  RecrodMetadata meta = f.get(); // blocking
} catch (ExecutionException ex) { ... }

Future.get(); 해당 메서드는 블로킹 되기에 배치 효과가 떨어진다. (처리량 저하)

처리량이 낮아도 되는 경우에만 사용한다.

전송 결과 확인: Callback


package org.apache.kafka.clients.producer;

public interface Callback {
  void onCompletion(RecordMetadata metadata, Exception exception);
}

사용 예시


producer.send(new ProducerRecord<>("topic", "value"), new Callback() {
  @Override
  public void onCompletion(RecordMetadata metadata, Exception ex) {... }
});

함수형 인터페이스로 Future.get(); 메서드와 다르게 논블로킹 방식으로 처리량 저하가 없다. (배치가 쌓이지 않음)

전송 보장과 ack

프로듀서가 보낸 메시지가 성공적으로 잘 도착했는지 메시지 전송 확인 방식에 대해 지정할 수 있다. 브로커는 메시지의 복제와 관리(고가용성, 데이터 내구성)를 위해 리더와 팔로워를 사용하는데 팔로워와 리더에 메시지가 잘 저장되었는지 ack 를 통해 설정하여 메시지 전송에 대한 보장을 처리할 수 있다.