kafka指定消费1分钟前的数据实现延迟消费 c++ 版本

要实现消费1分钟前的数据，可以使用Kafka的offsetsForTimes API。该API可以返回一个topic partition最早时间戳大于或等于指定时间戳的offset。在使用该API时，需要注意以下几点：

1. 时间戳需要转换为以毫秒为单位的Unix时间戳。
2. 需要先消费到最新的消息，然后再使用offsetsForTimes API获取指定时间戳的offset，并从该offset开始消费。

下面是一个简单的C++代码示例，用于实现消费1分钟前的数据：

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

using namespace std;

int64_t get_offset_for_time(RdKafka::KafkaConsumer *consumer, const string &topic, int partition, int64_t timestamp_ms) {
    RdKafka::TopicPartition tp(topic, partition, timestamp_ms);
    vector<RdKafka::TopicPartition*> partitions {&tp};
    consumer->offsetsForTimes(partitions, 1000); // 1 second timeout
    int64_t offset = partitions[0]->offset;
    delete partitions[0];
    return offset;
}

int main() {
    // Kafka configuration
    RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
    conf->set("metadata.broker.list", "localhost:9092", errstr);
    conf->set("group.id", "my-group", errstr);

    // Kafka consumer
    RdKafka::KafkaConsumer *consumer = RdKafka::KafkaConsumer::create(conf, errstr);
    consumer->subscribe({ "my-topic" });

    // Consume messages until 1 minute ago
    int64_t timestamp_ms = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now() - std::chrono::minutes(1)).count();
    int64_t offset = get_offset_for_time(consumer, "my-topic", 0, timestamp_ms);
    consumer->seek("my-topic", 0, offset);
    while (true) {
        RdKafka::Message *msg = consumer->consume(1000);
        if (msg && msg->err() == RdKafka::ERR_NO_ERROR) {
            // Handle message
        }
        delete msg;
    }

    // Cleanup
    consumer->close();
    delete consumer;
    delete conf;

    return 0;
}

在上面的代码中，我们首先创建了一个Kafka配置和一个Kafka消费者。然后，我们使用`get_offset_for_time`函数获取1分钟前的offset，并从该offset开始消费消息。

在`get_offset_for_time`函数中，我们创建了一个`TopicPartition`对象，该对象包含了指定分区和时间戳的信息。然后，我们将该对象传递给`offsetsForTimes`方法，并设置了1秒的超时时间。该方法返回一个包含了查询结果的`TopicPartition`对象的vector，我们从vector中获取offset，并删除`TopicPartition`对象。

最后，我们在主函数中使用`consume`方法从指定offset开始消费消息。请注意，上面的代码仅包含了一个分区的消费，如需消费多个分区，请参考RdKafka C++ API文档。