Spring Kafka: ReplyingKafkaTemplate回复的C++生产者

#include <iostream>
#include <string>
#include <librdkafka/rdkafkacpp.h>

class KafkaProducer {
public:
    KafkaProducer(const std::string& broker, const std::string& topic) {
        std::string errstr;
        conf_ = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
        conf_->set("bootstrap.servers", broker, errstr);
        conf_->set("message.send.max.retries", "3", errstr);
        producer_ = RdKafka::Producer::create(conf_, errstr);
        if (!producer_) {
            std::cerr << "Failed to create producer: " << errstr << std::endl;
            exit(1);
        }
        topic_ = RdKafka::Topic::create(producer_, topic, nullptr, errstr);
        if (!topic_) {
            std::cerr << "Failed to create topic: " << errstr << std::endl;
            exit(1);
        }
    }

    void produce(const std::string& message) {
        RdKafka::ErrorCode resp = producer_->produce(
            topic_, RdKafka::Topic::PARTITION_UA,
            RdKafka::Producer::RK_MSG_COPY,
            const_cast<char*>(message.c_str()), message.size(),
            nullptr, 0, nullptr, nullptr);

        if (resp != RdKafka::ERR_NO_ERROR) {
            std::cerr << "Failed to produce message: " << RdKafka::err2str(resp) << std::endl;
        } else {
            producer_->poll(0);
        }
    }

    ~KafkaProducer() {
        producer_->flush(1000);
        delete topic_;
        delete producer_;
        delete conf_;
    }

private:
    RdKafka::Conf* conf_;
    RdKafka::Producer* producer_;
    RdKafka::Topic* topic_;
};

int main() {
    KafkaProducer producer("localhost:9092", "test_topic");
    producer.produce("Hello, Kafka!");
    return 0;
}