C++ 클라이언트 사용

C++ 클라이언트 사용 (C++ client use)

C++ 클라이언트로 프로듀서와 컨슈머를 만들어 메시지를 보내고 받는 방법을 살펴볼게요. 프로듀서와 컨슈머 모두 동기(blocking) 방식과 비동기(non-blocking) 방식을 지원하니, 처리량과 코드 단순함 사이에서 상황에 맞게 고르면 돼요.

출처: 문서

본문

프로듀서 만들기 (Create a producer)

Pulsar를 프로듀서로 사용하려면 C++ 클라이언트에 프로듀서를 만들어야 해요. 프로듀서를 사용하는 두 가지 주요 방식이 있어요.

  • 블로킹(Blocking) 방식: send를 호출할 때마다 브로커의 ack를 기다려요.
  • 비차단 비동기(Non-blocking asynchronous) 방식: send 대신 sendAsync를 호출하고, 브로커로부터 ack를 받았을 때 실행될 callback을 제공해요.

단순 블로킹 예시 (Simple blocking example)

이 예시는 블로킹 방식으로 100개의 메시지를 보내요. 단순하지만, 다음 메시지를 보내기 전에 각각의 ack가 돌아오기를 기다리므로 높은 처리량을 내지는 못해요.

#include <pulsar/Client.h>
#include <thread>

using namespace pulsar;

int main() {
    Client client("pulsar://localhost:6650");

    Producer producer;

    Result result = client.createProducer("persistent://public/default/my-topic", producer);
    if (result != ResultOk) {
        std::cout << "Error creating producer: " << result << std::endl;
        return -1;
    }

    // Send 100 messages synchronously
    int ctr = 0;
    while (ctr < 100) {
        std::string content = "msg" + std::to_string(ctr);
        Message msg = MessageBuilder().setContent(content).setProperty("x", "1").build();
        Result result = producer.send(msg);
        if (result != ResultOk) {
            std::cout << "The message " << content << " could not be sent, received code: " << result << std::endl;
        } else {
            std::cout << "The message " << content << " sent successfully" << std::endl;
        }

        std::this_thread::sleep_for(std::chrono::milliseconds(100));
        ctr++;
    }

    std::cout << "Finished producing synchronously!" << std::endl;

    client.close();
    return 0;
}

비차단 예시 (Non-blocking example)

이 예시는 send 대신 sendAsync를 호출하는 비차단 방식으로 100개의 메시지를 보내요. 이렇게 하면 프로듀서가 한 번에 여러 메시지를 전송 중(in-flight)인 상태로 둘 수 있어 처리량이 높아져요.

프로듀서 구성 blockIfQueueFull은 발신 요청을 위한 내부 큐가 가득 찼을 때 ResultProducerQueueIsFull 에러를 피하는 데 유용해요. 내부 큐가 가득 차면 sendAsync가 블로킹이 되어 코드를 더 단순하게 만들 수 있어요. 이 구성이 없으면 결과 코드 ResultProducerQueueIsFull이 callback에 전달돼요. 그 코드를 어떻게 처리할지(재시도, 버리기 등) 직접 결정해야 해요.

#include <pulsar/Client.h>
#include <thread>
#include <atomic>

using namespace pulsar;

std::atomic<uint32_t> acksReceived;

void callback(Result code, const MessageId& msgId, std::string msgContent) {
    // message processing logic here
    std::cout << "Received ack for msg: " << msgContent << " with code: "
        << code << " -- MsgID: " << msgId << std::endl;
    acksReceived++;
}

int main() {
    Client client("pulsar://localhost:6650");

    ProducerConfiguration producerConf;
    producerConf.setBlockIfQueueFull(true);
    Producer producer;
    Result result = client.createProducer("persistent://public/default/my-topic",
                                          producerConf, producer);
    if (result != ResultOk) {
        std::cout << "Error creating producer: " << result << std::endl;
        return -1;
    }

    // Send 100 messages asynchronously
    int ctr = 0;
    while (ctr < 100) {
        std::string content = "msg" + std::to_string(ctr);
        Message msg = MessageBuilder().setContent(content).setProperty("x", "1").build();
        producer.sendAsync(msg, std::bind(callback,
                                          std::placeholders::_1, std::placeholders::_2, content));

        std::this_thread::sleep_for(std::chrono::milliseconds(100));
        ctr++;
    }

    // wait for 100 messages to be acked
    while (acksReceived < 100) {
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }

    std::cout << "Finished producing asynchronously!" << std::endl;

    client.close();
    return 0;
}

파티셔닝 토픽과 지연 프로듀서 (Partitioned topics and lazy producers)

Pulsar 토픽을 확장할 때 수백 개의 파티션을 갖도록 구성할 수 있어요. 마찬가지로 프로듀서도 수백, 심지어 수천 개로 확장했을 수 있죠. 파티셔닝 토픽에 프로듀서를 만들면 내부적으로 파티션마다 내부 프로듀서를 하나씩 만들고, 각각이 브로커와 통신해야 하므로 Pulsar 브로커에 부담이 될 수 있어요. 파티션이 1000개이고 프로듀서가 1000개인 토픽에 대해, 프로듀서 애플리케이션 전반에 걸쳐 1,000,000개의 내부 프로듀서가 만들어지고, 각각이 어느 브로커에 연결해야 하는지 알아내고 연결 핸드셰이크를 수행하기 위해 브로커와 통신해야 하죠.

다음과 같이 하면 많은 파티션과 많은 프로듀서가 결합되어 생기는 부하를 줄일 수 있어요.

  • SinglePartition 파티션 라우팅 모드 사용(모든 메시지가 무작위로 선택된 단일 파티션에만 보내지도록 보장)
  • 키가 없는 메시지 사용(메시지에 키가 있으면 키의 해시에 따라 라우팅되므로 메시지가 여러 파티션으로 보내지게 돼요)
  • 지연(lazy) 프로듀서 사용(메시지가 파티션으로 라우팅되어야 할 때만 내부 프로듀서를 생성하도록 보장)

위 예시의 경우, 1000개의 프로듀서 앱에 걸쳐 퍼져 있는 내부 프로듀서 수를 1,000,000개에서 1000개로 줄이게 돼요.

첫 메시지 전송에 추가 지연이 있을 수 있다는 점을 참고하세요. 낮은 send timeout을 설정했다면, 초기 연결 핸드셰이크 완료가 느리면 그 timeout에 도달할 수 있어요.

ProducerConfiguration producerConf;
producerConf.setPartitionsRoutingMode(ProducerConfiguration::UseSinglePartition);
producerConf.setLazyStartPartitionedProducers(true);

컨슈머 만들기 (Create a consumer)

Pulsar를 컨슈머로 사용하려면 C++ 클라이언트에 컨슈머를 만들어야 해요. 컨슈머를 사용하는 두 가지 주요 방식이 있어요.

  • 블로킹(Blocking) 방식: receive(msg)를 동기적으로 호출.
  • 비차단(이벤트 기반) 방식: 메시지 리스너 사용.

블로킹 예시 (Blocking example)

이 접근 방식의 장점은 코드가 가장 단순하다는 거예요. 메시지를 받을 때까지 블록하는 receive(msg)를 계속 호출하면 돼요. 이 예시는 가장 이른(earliest) 오프셋에서 구독을 시작하고 100개의 메시지를 소비해요.

#include <pulsar/Client.h>

using namespace pulsar;

int main() {
    Client client("pulsar://localhost:6650");

    Consumer consumer;
    ConsumerConfiguration config;
    config.setSubscriptionInitialPosition(InitialPositionEarliest);
    Result result = client.subscribe("persistent://public/default/my-topic", "consumer-1", config, consumer);
    if (result != ResultOk) {
        std::cout << "Failed to subscribe: " << result << std::endl;
        return -1;
    }

    Message msg;
    int ctr = 0;
    // consume 100 messages
    while (ctr < 100) {
        consumer.receive(msg);
        std::cout << "Received: " << msg
            << "  with payload '" << msg.getDataAsString() << "'" << std::endl;

        consumer.acknowledge(msg);
        ctr++;
    }

    std::cout << "Finished consuming synchronously!" << std::endl;

    client.close();
    return 0;
}

더 알아보기 (Learn more)