C++와 Kafka로 대규모 스트리밍 데이터 처리 파이프라인 구축하기

sudo apt update
sudo apt install librdkafka-dev

git clone https://github.com/edenhill/librdkafka.git
cd librdkafka
./configure
make
sudo make install

find_package(PkgConfig REQUIRED)
pkg_check_modules(RDKAFKA REQUIRED librdkafka)

add_executable(my_kafka_app main.cpp)
target_link_libraries(my_kafka_app PRIVATE ${RDKAFKA_LIBRARIES})

# Zookeeper 실행
bin/zookeeper-server-start.sh config/zookeeper.properties

# Kafka 브로커 실행
bin/kafka-server-start.sh config/server.properties

sudo apt install librdkafka-dev  # Ubuntu
brew install librdkafka          # macOS

git clone https://github.com/mfontanini/cppkafka.git
cd cppkafka
mkdir build && cd build
cmake ..
make
sudo make install

sudo apt install librdkafka-dev  # Ubuntu
brew install librdkafka          # macOS

#include <iostream>
#include <rdkafka.h>

// Kafka 메시지 전송 콜백 함수
void delivery_report_cb(rd_kafka_t *rk, const rd_kafka_message_t *rkmessage, void *opaque) {
    if (rkmessage->err) {
        std::cerr << "❌ 메시지 전송 실패: " << rd_kafka_err2str(rkmessage->err) << std::endl;
    } else {
        std::cout << "✅ 메시지 전송 성공: " << static_cast<char*>(rkmessage->payload) << std::endl;
    }
}

int main() {
    const std::string brokers = "localhost:9092";  // Kafka 브로커 주소
    const std::string topic_name = "test_topic";   // 전송할 토픽 이름

    // Kafka 설정 객체 생성
    rd_kafka_conf_t *conf = rd_kafka_conf_new();
    rd_kafka_conf_set(conf, "bootstrap.servers", brokers.c_str(), NULL, 0);

    // 프로듀서 생성
    rd_kafka_t *producer = rd_kafka_new(RD_KAFKA_PRODUCER, conf, NULL, 0);
    if (!producer) {
        std::cerr << "❌ Kafka 프로듀서 생성 실패!" << std::endl;
        return 1;
    }

    // 메시지 전송 루프
    for (int i = 1; i <= 5; ++i) {
        std::string message = "Kafka 메시지 " + std::to_string(i);

        rd_kafka_resp_err_t err = rd_kafka_producev(
            producer,
            RD_KAFKA_V_TOPIC(topic_name.c_str()),
            RD_KAFKA_V_VALUE(message.c_str(), message.size()),
            RD_KAFKA_V_END
        );

        if (err) {
            std::cerr << "❌ 메시지 전송 실패: " << rd_kafka_err2str(err) << std::endl;
        } else {
            std::cout << "📤 메시지 전송됨: " << message << std::endl;
        }

        // Kafka 내부 큐를 비우기 위해 플러시 수행
        rd_kafka_flush(producer, 5000);
    }

    // 프로듀서 종료
    rd_kafka_destroy(producer);

    return 0;
}

# Zookeeper 실행
bin/zookeeper-server-start.sh config/zookeeper.properties

# Kafka 브로커 실행
bin/kafka-server-start.sh config/server.properties

# 토픽 생성
bin/kafka-topics.sh --create --topic test_topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

g++ -o kafka_producer kafka_producer.cpp -lrdkafka
./kafka_producer

bin/kafka-console-consumer.sh --topic test_topic --from-beginning --bootstrap-server localhost:9092

rd_kafka_conf_set(conf, "linger.ms", "5", NULL, 0);
rd_kafka_conf_set(conf, "batch.size", "16384", NULL, 0);
rd_kafka_conf_set(conf, "acks", "all", NULL, 0);
rd_kafka_conf_set(conf, "compression.type", "snappy", NULL, 0);

sudo apt install librdkafka-dev  # Ubuntu
brew install librdkafka          # macOS

#include <iostream>
#include <csignal>
#include <rdkafka.h>

bool run = true;

// SIGINT 또는 SIGTERM 신호를 처리하는 핸들러
void signal_handler(int signal) {
    std::cout << "🛑 종료 신호 수신. 컨슈머를 종료합니다..." << std::endl;
    run = false;
}

int main() {
    const std::string brokers = "localhost:9092";  // Kafka 브로커 주소
    const std::string topic_name = "test_topic";   // 구독할 토픽
    const std::string group_id = "cpp_consumer_group";  // 컨슈머 그룹 ID

    // SIGINT(Ctrl+C) 신호 처리 등록
    signal(SIGINT, signal_handler);
    signal(SIGTERM, signal_handler);

    // Kafka 설정 객체 생성
    rd_kafka_conf_t *conf = rd_kafka_conf_new();
    rd_kafka_conf_set(conf, "bootstrap.servers", brokers.c_str(), NULL, 0);
    rd_kafka_conf_set(conf, "group.id", group_id.c_str(), NULL, 0);
    rd_kafka_conf_set(conf, "auto.offset.reset", "earliest", NULL, 0);

    // Kafka 컨슈머 생성
    rd_kafka_t *consumer = rd_kafka_new(RD_KAFKA_CONSUMER, conf, NULL, 0);
    if (!consumer) {
        std::cerr << "❌ Kafka 컨슈머 생성 실패!" << std::endl;
        return 1;
    }

    // 토픽 구독 설정
    rd_kafka_poll_set_consumer(consumer);
    rd_kafka_topic_partition_list_t *topics = rd_kafka_topic_partition_list_new(1);
    rd_kafka_topic_partition_list_add(topics, topic_name.c_str(), -1);
    rd_kafka_subscribe(consumer, topics);
    rd_kafka_topic_partition_list_destroy(topics);

    std::cout << "📡 Kafka 컨슈머가 " << topic_name << " 토픽을 구독 중..." << std::endl;

    // 메시지 수신 루프
    while (run) {
        rd_kafka_message_t *msg = rd_kafka_consumer_poll(consumer, 1000);
        if (msg) {
            if (msg->err) {
                std::cerr << "⚠️ 오류 발생: " << rd_kafka_message_errstr(msg) << std::endl;
            } else {
                std::cout << "📥 메시지 수신: " << static_cast<char*>(msg->payload) << std::endl;
            }
            rd_kafka_message_destroy(msg);
        }
    }

    // 컨슈머 종료
    rd_kafka_consumer_close(consumer);
    rd_kafka_destroy(consumer);

    return 0;
}

# Zookeeper 실행
bin/zookeeper-server-start.sh config/zookeeper.properties

# Kafka 브로커 실행
bin/kafka-server-start.sh config/server.properties

bin/kafka-console-producer.sh --topic test_topic --bootstrap-server localhost:9092
> 메시지 1
> 메시지 2
> 메시지 3

g++ -o kafka_consumer kafka_consumer.cpp -lrdkafka
./kafka_consumer

📡 Kafka 컨슈머가 test_topic 토픽을 구독 중...
📥 메시지 수신: 메시지 1
📥 메시지 수신: 메시지 2
📥 메시지 수신: 메시지 3

rd_kafka_conf_set(conf, "fetch.min.bytes", "50000", NULL, 0);
rd_kafka_conf_set(conf, "fetch.wait.max.ms", "100", NULL, 0);
rd_kafka_conf_set(conf, "enable.auto.commit", "false", NULL, 0);
rd_kafka_conf_set(conf, "session.timeout.ms", "60000", NULL, 0);

rd_kafka_conf_set(conf, "batch.num.messages", "10000", NULL, 0);
rd_kafka_conf_set(conf, "linger.ms", "10", NULL, 0);

rd_kafka_conf_set(conf, "compression.type", "snappy", NULL, 0);

#include <iostream>
#include <thread>
#include <vector>
#include <rdkafka.h>

void consume_messages(const std::string &topic, int partition) {
    rd_kafka_conf_t *conf = rd_kafka_conf_new();
    rd_kafka_conf_set(conf, "group.id", "consumer_group", NULL, 0);
    rd_kafka_t *consumer = rd_kafka_new(RD_KAFKA_CONSUMER, conf, NULL, 0);

    rd_kafka_topic_partition_list_t *topics = rd_kafka_topic_partition_list_new(1);
    rd_kafka_topic_partition_list_add(topics, topic.c_str(), partition);
    rd_kafka_assign(consumer, topics);

    while (true) {
        rd_kafka_message_t *msg = rd_kafka_consumer_poll(consumer, 1000);
        if (msg) {
            std::cout << "📥 [Partition " << partition << "] 메시지 수신: " 
                      << static_cast<char*>(msg->payload) << std::endl;
            rd_kafka_message_destroy(msg);
        }
    }
}

int main() {
    const std::string topic = "test_topic";
    const int num_threads = 4;

    std::vector<std::thread> threads;
    for (int i = 0; i < num_threads; ++i) {
        threads.emplace_back(consume_messages, topic, i);
    }

    for (auto &t : threads) {
        t.join();
    }

    return 0;
}

rd_kafka_conf_set(conf, "enable.auto.commit", "false", NULL, 0);

rd_kafka_commit(consumer, NULL, 0);

num.partitions=4
replication.factor=3

bin/kafka-run-class.sh kafka.tools.ConsumerPerformance --topic test_topic --bootstrap-server localhost:9092

rd_kafka_conf_set(conf, "message.send.max.retries", "5", NULL, 0);
rd_kafka_conf_set(conf, "retry.backoff.ms", "100", NULL, 0);

#include <iostream>
#include <rdkafka.h>

// Kafka 메시지 전송 실패 시 로깅
void delivery_report_cb(rd_kafka_t *rk, const rd_kafka_message_t *rkmessage, void *opaque) {
    if (rkmessage->err) {
        std::cerr << "❌ 메시지 전송 실패: " << rd_kafka_err2str(rkmessage->err) << std::endl;
    } else {
        std::cout << "✅ 메시지 전송 성공: " << static_cast<char*>(rkmessage->payload) << std::endl;
    }
}

int main() {
    const std::string brokers = "localhost:9092";
    const std::string topic_name = "test_topic";

    rd_kafka_conf_t *conf = rd_kafka_conf_new();
    rd_kafka_conf_set(conf, "bootstrap.servers", brokers.c_str(), NULL, 0);
    rd_kafka_conf_set(conf, "message.send.max.retries", "5", NULL, 0);  // 전송 재시도 횟수
    rd_kafka_conf_set(conf, "retry.backoff.ms", "100", NULL, 0);  // 재시도 간격

    rd_kafka_t *producer = rd_kafka_new(RD_KAFKA_PRODUCER, conf, NULL, 0);

    if (!producer) {
        std::cerr << "❌ Kafka 프로듀서 생성 실패!" << std::endl;
        return 1;
    }

    std::string message = "Kafka 오류 처리 테스트 메시지";
    rd_kafka_resp_err_t err = rd_kafka_producev(
        producer,
        RD_KAFKA_V_TOPIC(topic_name.c_str()),
        RD_KAFKA_V_VALUE(message.c_str(), message.size()),
        RD_KAFKA_V_END
    );

    if (err) {
        std::cerr << "❌ 메시지 전송 실패: " << rd_kafka_err2str(err) << std::endl;
    } else {
        std::cout << "📤 메시지 전송됨: " << message << std::endl;
    }

    rd_kafka_flush(producer, 5000);
    rd_kafka_destroy(producer);

    return 0;
}

#include <iostream>
#include <csignal>
#include <rdkafka.h>

bool run = true;

// 종료 신호 핸들러
void signal_handler(int signal) {
    std::cout << "🛑 종료 신호 수신. 컨슈머 종료..." << std::endl;
    run = false;
}

int main() {
    const std::string brokers = "localhost:9092";
    const std::string topic_name = "test_topic";
    const std::string group_id = "cpp_consumer_group";

    signal(SIGINT, signal_handler);
    signal(SIGTERM, signal_handler);

    rd_kafka_conf_t *conf = rd_kafka_conf_new();
    rd_kafka_conf_set(conf, "bootstrap.servers", brokers.c_str(), NULL, 0);
    rd_kafka_conf_set(conf, "group.id", group_id.c_str(), NULL, 0);
    rd_kafka_conf_set(conf, "enable.auto.commit", "false", NULL, 0);  // 수동 커밋 설정

    rd_kafka_t *consumer = rd_kafka_new(RD_KAFKA_CONSUMER, conf, NULL, 0);
    if (!consumer) {
        std::cerr << "❌ Kafka 컨슈머 생성 실패!" << std::endl;
        return 1;
    }

    rd_kafka_poll_set_consumer(consumer);
    rd_kafka_topic_partition_list_t *topics = rd_kafka_topic_partition_list_new(1);
    rd_kafka_topic_partition_list_add(topics, topic_name.c_str(), -1);
    rd_kafka_subscribe(consumer, topics);
    rd_kafka_topic_partition_list_destroy(topics);

    std::cout << "📡 Kafka 컨슈머가 " << topic_name << " 토픽을 구독 중..." << std::endl;

    while (run) {
        rd_kafka_message_t *msg = rd_kafka_consumer_poll(consumer, 1000);
        if (msg) {
            if (msg->err) {
                std::cerr << "⚠️ 오류 발생: " << rd_kafka_message_errstr(msg) << std::endl;
                if (msg->err == RD_KAFKA_RESP_ERR__TRANSPORT) {
                    std::cerr << "🔄 네트워크 문제 발생, 재연결 시도 중..." << std::endl;
                }
            } else {
                std::cout << "📥 메시지 수신: " << static_cast<char*>(msg->payload) << std::endl;
                rd_kafka_commit(consumer, NULL, 0);  // 메시지 소비 후 수동 오프셋 커밋
            }
            rd_kafka_message_destroy(msg);
        }
    }

    rd_kafka_consumer_close(consumer);
    rd_kafka_destroy(consumer);

    return 0;
}

[데이터 소스] → [Kafka 프로듀서] → [Kafka 브로커] → [Kafka 컨슈머] → [데이터베이스/분석 시스템]

#include <iostream>
#include <rdkafka.h>
#include <random>
#include <chrono>
#include <thread>

// 센서 데이터를 생성하는 함수
std::string generate_sensor_data() {
    std::random_device rd;
    std::mt19937 gen(rd());
    std::uniform_real_distribution<> temp_dist(15.0, 35.0);
    std::uniform_real_distribution<> humidity_dist(30.0, 80.0);

    double temperature = temp_dist(gen);
    double humidity = humidity_dist(gen);

    return "temperature:" + std::to_string(temperature) + ",humidity:" + std::to_string(humidity);
}

int main() {
    const std::string brokers = "localhost:9092";
    const std::string topic_name = "sensor_data";

    rd_kafka_conf_t *conf = rd_kafka_conf_new();
    rd_kafka_conf_set(conf, "bootstrap.servers", brokers.c_str(), NULL, 0);

    rd_kafka_t *producer = rd_kafka_new(RD_KAFKA_PRODUCER, conf, NULL, 0);
    if (!producer) {
        std::cerr << "❌ Kafka 프로듀서 생성 실패!" << std::endl;
        return 1;
    }

    while (true) {
        std::string sensor_data = generate_sensor_data();

        rd_kafka_resp_err_t err = rd_kafka_producev(
            producer,
            RD_KAFKA_V_TOPIC(topic_name.c_str()),
            RD_KAFKA_V_VALUE(sensor_data.c_str(), sensor_data.size()),
            RD_KAFKA_V_END
        );

        if (err) {
            std::cerr << "❌ 메시지 전송 실패: " << rd_kafka_err2str(err) << std::endl;
        } else {
            std::cout << "📤 센서 데이터 전송됨: " << sensor_data << std::endl;
        }

        rd_kafka_flush(producer, 1000);
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }

    rd_kafka_destroy(producer);
    return 0;
}

#include <iostream>
#include <fstream>
#include <rdkafka.h>

bool run = true;

void signal_handler(int signal) {
    std::cout << "🛑 종료 신호 수신. 컨슈머 종료..." << std::endl;
    run = false;
}

int main() {
    const std::string brokers = "localhost:9092";
    const std::string topic_name = "sensor_data";
    const std::string group_id = "sensor_consumer_group";
    std::ofstream log_file("sensor_log.txt");

    rd_kafka_conf_t *conf = rd_kafka_conf_new();
    rd_kafka_conf_set(conf, "bootstrap.servers", brokers.c_str(), NULL, 0);
    rd_kafka_conf_set(conf, "group.id", group_id.c_str(), NULL, 0);
    rd_kafka_conf_set(conf, "auto.offset.reset", "earliest", NULL, 0);

    rd_kafka_t *consumer = rd_kafka_new(RD_KAFKA_CONSUMER, conf, NULL, 0);
    rd_kafka_poll_set_consumer(consumer);

    rd_kafka_topic_partition_list_t *topics = rd_kafka_topic_partition_list_new(1);
    rd_kafka_topic_partition_list_add(topics, topic_name.c_str(), -1);
    rd_kafka_subscribe(consumer, topics);
    rd_kafka_topic_partition_list_destroy(topics);

    while (run) {
        rd_kafka_message_t *msg = rd_kafka_consumer_poll(consumer, 1000);
        if (msg && !msg->err) {
            std::string received_data = static_cast<char*>(msg->payload);
            std::cout << "📥 수신된 센서 데이터: " << received_data << std::endl;
            log_file << received_data << std::endl;
            log_file.flush();
        }
        rd_kafka_message_destroy(msg);
    }

    log_file.close();
    rd_kafka_consumer_close(consumer);
    rd_kafka_destroy(consumer);

    return 0;
}