c++kafka实现延迟消费的代码

以下是一个使用librdkafka库实现延迟消费的C++代码示例：

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

class ExampleDeliveryReportCb : public RdKafka::DeliveryReportCb {
public:
    void dr_cb(RdKafka::Message &message) {
        if (message.err()) {
            std::cerr << "Message delivery failed: " << message.errstr() << std::endl;
        } else {
            std::cout << "Message delivered to topic " << message.topic_name() << " [" << message.partition() << "] at offset " << message.offset() << std::endl;
        }
    }
};

int main() {
    std::string brokers = "localhost:9092";
    std::string topic = "test";
    int32_t partition = RdKafka::Topic::PARTITION_UA;
    std::string message_str = "Hello, Kafka!";
    int32_t delay_ms = 5000; // 5秒延迟

    ExampleDeliveryReportCb ex_dr_cb;
    RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
    conf->set("metadata.broker.list", brokers, errstr);
    RdKafka::Producer *producer = RdKafka::Producer::create(conf, errstr);
    if (!producer) {
        std::cerr << "Failed to create producer: " << errstr << std::endl;
        exit(1);
    }

    RdKafka::Topic *topic_obj = RdKafka::Topic::create(producer, topic, NULL, errstr);
    if (!topic_obj) {
        std::cerr << "Failed to create topic: " << errstr << std::endl;
        exit(1);
    }

    RdKafka::Headers *headers = RdKafka::Headers::create();
    headers->add("X-Message-Delay", std::to_string(delay_ms), errstr);

    RdKafka::ErrorCode resp = producer->produce(topic_obj, partition, RdKafka::Producer::RK_MSG_COPY, const_cast<char *>(message_str.c_str()), message_str.size(), NULL, 0, (void *)headers, NULL);
    if (resp != RdKafka::ERR_NO_ERROR) {
        std::cerr << "Failed to produce message: " << RdKafka::err2str(resp) << std::endl;
    } else {
        std::cout << "Message produced to topic " << topic << std::endl;
    }

    producer->poll(0);

    RdKafka::Topic::destroy(topic_obj);
    delete headers;
    delete producer;
    delete conf;

    // 等待消息被消费
    std::this_thread::sleep_for(std::chrono::milliseconds(delay_ms));

    // 创建消费者
    conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
    conf->set("metadata.broker.list", brokers, errstr);
    RdKafka::Consumer *consumer = RdKafka::Consumer::create(conf, errstr);
    if (!consumer) {
        std::cerr << "Failed to create consumer: " << errstr << std::endl;
        exit(1);
    }

    RdKafka::TopicPartition *topic_partition = RdKafka::TopicPartition::create(topic, partition);
    topic_partition->set_offset(RdKafka::Topic::OFFSET_END);

    RdKafka::ErrorCode resp = consumer->assign({ topic_partition });
    if (resp != RdKafka::ERR_NO_ERROR) {
        std::cerr << "Failed to assign partition: " << RdKafka::err2str(resp) << std::endl;
        exit(1);
    }

    while (true) {
        RdKafka::Message *message = consumer->consume(1000);
        if (message) {
            if (message->err() == RdKafka::ERR_NO_ERROR) {
                std::cout << "Message consumed from topic " << message->topic_name() << " [" << message->partition() << "] at offset " << message->offset() << ": " << std::string(static_cast<char *>(message->payload()), message->len()) << std::endl;
                break; // 只消费一条消息
            } else if (message->err() == RdKafka::ERR__TIMED_OUT) {
                std::cout << "No message received within timeout." << std::endl;
            } else {
                std::cerr << "Failed to consume message: " << message->errstr() << std::endl;
            }
            delete message;
        }
    }

    consumer->close();
    delete consumer;
    delete conf;
    delete topic_partition;

    return 0;
}

该代码中，我们首先使用生产者将消息发送到Kafka，并设置了一个“X-Message-Delay”头部，指定了延迟消费的时间。然后，我们等待了指定的时间，再创建消费者，并消费该消息。

请注意，该代码仅演示了如何实现延迟消费，实际生产环境中，您需要对代码进行更多的优化和错误处理。