KafkaProducer

💡 이번 포스팅에서는 KafkaProducer 객체에 대해 알아보도록 하겠습니다.

 

kafka의 producer에 대해 궁금하신 분들은 아래 producer에 대해서 정리한 포스팅이 있으니 참고하시면 좋을 것 같습니다.

 

Producer

 

kafka를 사용하기 위해서는 아래와 같이 의존성을 추가해야 합니다.

 

implementation 'org.apache.kafka:kafka-clients:3.1.2'

 

그리고 kafka-client에서 제공하는 KafkaProducer 객체를 통해 producer를 생성합니다.

이때, producer 설정 정보를 전달해줘야 하는데 Bootstrap-server, key-serializer, value-serializer 정보는 필수 값입니다.

Properties props = new Properties();

// bootstrap.servers, key.serializer.class, value.serializer.class
props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.111.133:9092");
props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

// KafkaProducer 객체 생성
KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(props)

 

그리고 ProducerRecord 객체를 생성하여 프로듀서에 전달할 메시지를 생성합니다.

// ProducerRecord 객체 생성
ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topicName, "hello world");

기본적으로는 토픽, 값은 필수값이며 key, partition은 선택에 따라 설정할 수도 있습니다.

📢 producer 포스팅 참고

 

이를 확인해보기 위해 ProducerRecord 객체의 생성자를 확인해 보면 다양한 생성자들이 오버로딩 되어 있는 것을 확인할 수 있습니다.

 

프로듀서로 보낼 메시지를 생성했으면 이제 send() 메서드를 통해 메시지를 카프카 브로커에 전달할 수 있습니다.

// KafkaProducer 메시지 전송
kafkaProducer.send(producerRecord);​

 

send() 메서드는 기본적으로 비동기 방식으로 메시지가 전송되며 callback 과 get() 메서드등을 통해 동기/비동기 방식을 선택할 수 있습니다.

 

 

동기 방식

send() 메서드를 호출하면 기본적으로 Future<RecordMetadata> 타입이 반환됩니다.

Future 객체의 get() 메서드를 호출함으로써 프로듀서로 메시지를 전송하는 방식을 동기적으로 처리할 수 있습니다.

그리고 반환 타입은 RecordMetadata가 되며 반환된 RecordMetadata에 전달한 메시지의 토픽, 파티션, 오프셋과 같은 정보들이 담겨 있습니다.

try {
    RecordMetadata recordMetadata = kafkaProducer.send(producerRecord).get();
    logger.info("\n ##### record metadata received ##### \n");
    logger.info("partition : {}", recordMetadata.partition());
    logger.info("offset : {}", recordMetadata.offset());
    logger.info("timestamp : {}", recordMetadata.timestamp());
} catch (InterruptedException | ExecutionException e) {
    e.printStackTrace();
}

 

비동기 방식

send()는 기본적으로 비동기 방식이기 때문에 비동기 방식으로 메시지를 전달하고 싶은 경우에는 특별한 설정을 할 필요는 없지만, 메시지 전송 결과를 받아보고 싶은 경우에는 callback을 사용해야 합니다.

// KafkaProducer 메시지 전송 (비동기, callback 사용)
kafkaProducer.send(producerRecord, (metadata, exception) -> {
    if (exception == null) {
        logger.info("\n ##### record metadata received ##### \n");
        logger.info("partition : {}", metadata.partition());
        logger.info("offset : {}", metadata.offset());
        logger.info("timestamp : {}", metadata.timestamp());
    } else {
        logger.error("exception error from broker {}", exception.getMessage());
    }
});

callback을 사용하면 RecordMetadata, Exception이 파라미터로 전달되는데, 메시지 전송에 실패할 경우 Exception 객체에 예외 정보가 담겨있습니다.

따라서 개발자가 예외 처리를 하는 추가 작업이 필요하며, RecordMetadata에는 동기 방식과 마찬가지로 전송한 메시지의 관한 정보가 담겨 있습니다.

 

기본적으로는 lambda나 익명 클래스로 callback을 작성하는데 별도의 클래스로 작성하여 callback을 구현할 수도 있습니다.

public class CustomCallback implements Callback {

    private static final Logger logger = LoggerFactory.getLogger(CustomCallback.class.getName());

    private int seq;

    public CustomCallback(int seq) {
        this.seq = seq;
    }

    @Override
    public void onCompletion(RecordMetadata metadata, Exception exception) {
        if (exception == null) {
            logger.info("seq: {} partition: {}, offset: {},", this.seq, metadata.partition(), metadata.offset());
        } else {
            logger.error("exception error from broker {}", exception.getMessage());
        }
    }
}


// 사용
// KafkaProducer 메시지 전송 (비동기, callback 사용)
Callback callback = new CustomCallback(seq);
kafkaProducer.send(producerRecord, callback);

 

 

참고 자료

• <실전 카프카 개발부터 운영까지> - 고승범 지음
• <인프런 카프카 완벽 가이드 - 코어편> - 권철민