c++用kafka延迟队列来实现延迟消费

在C++中，可以使用librdkafka库来实现Kafka的延迟队列。下面是一个简单的示例代码：

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

int main() {
    // 创建Kafka配置
    RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
    conf->set("bootstrap.servers", "localhost:9092");

    // 创建生产者
    RdKafka::Producer *producer = RdKafka::Producer::create(conf, errstr);
    if (!producer) {
        std::cerr << "Failed to create producer: " << errstr << std::endl;
        exit(1);
    }

    // 创建延迟队列
    RdKafka::Headers *headers = RdKafka::Headers::create();
    headers->add("delay", "1000");

    // 发送消息到Kafka
    RdKafka::ErrorCode err = producer->produce(
        "my_topic", RdKafka::Topic::PARTITION_UA, 
        RdKafka::Producer::RK_MSG_COPY, 
        const_cast<char*>("my_message"), 
        strlen("my_message"),
        NULL, 0, 
        std::chrono::system_clock::now() + std::chrono::milliseconds(1000),
        headers);

    if (err != RdKafka::ERR_NO_ERROR) {
        std::cerr << "Failed to produce message: " << RdKafka::err2str(err) << std::endl;
    }

    // 等待一段时间
    std::this_thread::sleep_for(std::chrono::milliseconds(2000));

    // 关闭生产者
    delete producer;
    delete conf;
    delete headers;

    return 0;
}

在上面的示例代码中，我们使用了librdkafka库来创建Kafka生产者，并且使用了`RdKafka::Headers`来创建延迟队列。我们通过`headers->add("delay", "1000")`将延迟时间设置为1秒钟。我们还使用了`std::chrono::system_clock::now() + std::chrono::milliseconds(1000)`来设置消息的延迟时间。最后，我们使用`std::this_thread::sleep_for(std::chrono::milliseconds(2000))`等待2秒钟，以便让消息能够被消费者消费。

在消费者端，我们可以通过设置`auto.offset.reset`为`earliest`来消费已经过期的消息，例如：

#include <librdkafka/rdkafkacpp.h>

int main() {
    // 创建Kafka配置
    RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
    conf->set("bootstrap.servers", "localhost:9092");
    conf->set("group.id", "my_group");
    conf->set("auto.offset.reset", "earliest");

    // 创建消费者
    RdKafka::Consumer *consumer = RdKafka::Consumer::create(conf, errstr);
    if (!consumer) {
        std::cerr << "Failed to create consumer: " << errstr << std::endl;
        exit(1);
    }

    // 订阅主题
    RdKafka::Topic *topic = RdKafka::Topic::create(consumer, "my_topic", NULL, errstr);
    RdKafka::ErrorCode err = consumer->start(topic, 0, RdKafka::Topic::OFFSET_STORED);
    if (err != RdKafka::ERR_NO_ERROR) {
        std::cerr << "Failed to start consumer: " << RdKafka::err2str(err) << std::endl;
        exit(1);
    }

    // 消费消息
    while (true) {
        RdKafka::Message *message = consumer->consume(topic, 0, 1000);
        if (!message) {
            continue;
        }

        if (message->err() == RdKafka::ERR_NO_ERROR) {
            std::cout << "Received message: " << std::string(static_cast<char*>(message->payload()), message->len()) << std::endl;
        }

        delete message;
    }

    // 关闭消费者
    consumer->stop(topic, 0);
    delete topic;
    delete consumer;
    delete conf;

    return 0;
}

在上面的示例代码中，我们创建了一个Kafka消费者，并且通过设置`auto.offset.reset`为`earliest`来消费已经过期的消息。然后，我们订阅了`my_topic`主题，并在一个无限循环中使用`consumer->consume()`来消费消息。当我们收到一条消息时，我们打印出消息的内容。最后，我们使用`consumer->stop()`来关闭消费者。

总的来说，使用Kafka的延迟队列来实现延迟消费是一种非常方便的方式。通过使用librdkafka库，我们可以很容易地在C++中实现这个功能。