[Kafka] kafka produce 하기 (with C)

처음엔 그냥 파이썬으로 파일을 읽어서 카프카 메시지를 보내려 했는데 생각보다 느렸다.. 

카프카로 보내야하는 파일을 생성하는 코드가 c/c++로 되어있기도 하고 메시지 생성 즉시 바로 보내면 더 빨라질 것 같아 C로 작성해봤다.

 

1. librdkafka 패키지 다운로드

나는 centos 써서 아래와 같이 설치했다.

yum install librdkafka-devel

다른 운영체제는 아래 github 페이지 참고
https://github.com/confluentinc/librdkafka#installation

2. librdkafka 설치 위치 확인

설치 위치 확인하는 방법 참고: https://quackstudy.tistory.com/entry/CentOS-yum%EC%9C%BC%EB%A1%9C-%EC%84%A4%EC%B9%98%ED%95%9C-%ED%8C%A8%ED%82%A4%EC%A7%80-%EC%84%A4%EC%B9%98-%EA%B2%BD%EB%A1%9C-%ED%99%95%EC%9D%B8%ED%95%98%EB%8A%94-%EB%B0%A9%EB%B2%95

/usr/include/librdkafka/rdkafka.h
/usr/lib64/librdkafka.so

빌드할때에는 위 파일 사용 예정

3. C 코드 작성

참고한 코드:https://github.com/confluentinc/librdkafka/blob/master/examples/producer.c

3-0. librdkafka import

#include "rdkafka.h"

3-1. producer instance 생성

int
createKafkaProducer(rd_kafka_t **rk, const char *brokers){
	rd_kafka_conf_t *conf;
    char errstr[512];
    
    conf = rd_kafka_conf_new()
    

    /* kafka configuration 세팅 */
    if (rd_kafka_conf_set(conf, "bootstrap.servers", brokers, errstr, sizeof(errstr))!=RD_KAFKA_CONF_OK){
            printf("%s\n", errstr);
            return -1;
    }

    /* kafka producer instance 생성 */
    *rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr, sizeof(errstr));
    if(!rk){
            printf("Failed to create new producer: %s\n",errstr);
            return 1;
    }
    
    return 0;
  
 }

3-2. message produce

void
producetoKafka(rd_kafka_t *rk, const char *topic, const char *message){

        size_t message_len = strlen(message);
        rd_kafka_resp_err_t err;
retry:

        err = rd_kafka_producev(rk, RD_KAFKA_V_TOPIC(topic),
                                                        RD_KAFKA_V_MSGFLAGS(RD_KAFKA_MSG_F_COPY),
                                                        RD_KAFKA_V_VALUE(message, message_len),
                                                        RD_KAFKA_V_OPAQUE(NULL),
                                                        RD_KAFKA_V_END);

        if(err){
            fprintf(stderr, "Failed to produce to topic %s: %s\n", topic,  rd_kafka_err2str(err));
            
            /* local queue가 가득차면 produce 재시도 */
            if (err == RD_KAFKA_RESP_ERR__QUEUE_FULL) {
                fprintf(stderr, "poll...\n");
                rd_kafka_poll(rk,1000);
                goto retry;
            }
        } else {
                printf("Enqueued message (%zd bytes) for topic %s\n", message_len, topic);
        }

        rd_kafka_poll(rk, 0);

}

3-3. Destroy producer

void
destroyKafkaProducer(rd_kafka_t *rk){
    printf("Flushing final messages..\n");
    rd_kafka_flush(rk, 10*1000); //timeout value (ms단위)

    if(rd_kafka_outq_len(rk)>0){
        fprintf(stderr,"%d message(s) were not delivered\n", rd_kafka_outq_len(rk));
    }
    rd_kafka_destroy(rk);
}

3-4. Makefile 작성

2번에서 확인했던 헤더파일 디렉토리를 추가해주면 된다.

-I/usr/include/librdkafka
-lrdkafka

.so 파일 사용을 위해 LD_LIBRARY_PATH에도 경로 추가해줬다.

LD_LIBRARY_PATH=some/path....:/usr/lib64/pkgconfig

 

 

함수별 자세한 설명은  `rdkafka.h`파일에 주석 읽어보기